flakes

.landscape.yml

@@ -1,6 +1,8 @@
 doc-warnings: false
 test-warnings: false
 max-line-length: 79
+inherits:
+    - strictness_veryhigh
 uses:
     - celery
 autodetect: true
@@ -34,5 +36,6 @@ pylint:
         - lost-exception
         - dangerous-default-value
         - unused-argument
+        - protected-access
     options:
         exclude-protected: _reader, _writer, _popen, _sentinel_poll, _job, _is_alive, _write_to, _scheduled_for, _terminated, _accepted, _set_terminated, _payload, _cancel

celery/backends/async.py

@@ -218,7 +218,7 @@ class BaseResultConsumer(object):
         self.buckets = WeakKeyDictionary()
         self.drainer = drainers[detect_environment()](self)
 
-    def start(self):
+    def start(self, initial_task_id):
         raise NotImplementedError()
 
     def stop(self):

celery/backends/redis.py

@@ -23,11 +23,9 @@ from . import base
 
 try:
     import redis
-    from redis.exceptions import ConnectionError
     from kombu.transport.redis import get_redis_error_classes
 except ImportError:                 # pragma: no cover
     redis = None                    # noqa
-    ConnectionError = None          # noqa
     get_redis_error_classes = None  # noqa
 
 __all__ = ['RedisBackend']
@@ -142,7 +140,7 @@ class RedisBackend(base.BaseKeyValueStoreBackend, async.AsyncBackendMixin):
         )
 
     def _params_from_url(self, url, defaults):
-        scheme, host, port, user, password, path, query = _parse_url(url)
+        scheme, host, port, _, password, path, query = _parse_url(url)
         connparams = dict(
             defaults, **dictfilter({
                 'host': host, 'port': port, 'password': password,
@@ -236,7 +234,8 @@ class RedisBackend(base.BaseKeyValueStoreBackend, async.AsyncBackendMixin):
         options['task_id'] = group_id
         return header(*partial_args, **options or {})
 
-    def on_chord_part_return(self, request, state, result, propagate=None):
+    def on_chord_part_return(self, request, state, result,
+                             propagate=None, **kwargs):
         app = self.app
         tid, gid = request.id, request.group
         if not gid or not tid:

celery/backends/riak.py

@@ -66,7 +66,7 @@ class RiakBackend(KeyValueStoreBackend):
     #: default Riak server port (8087)
     port = 8087
 
-    # supports_autoexpire = False
+    _bucket = None
 
     def __init__(self, host=None, port=None, bucket_name=None, protocol=None,
                  url=None, *args, **kwargs):
@@ -78,9 +78,9 @@ class RiakBackend(KeyValueStoreBackend):
                 'You need to install the riak library to use the '
                 'Riak backend.')
 
-        uhost = uport = uname = upass = ubucket = None
+        uhost = uport = upass = ubucket = None
         if url:
-            uprot, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)
+            _, uhost, uport, _, upass, ubucket, _ = _parse_url(url)
             if ubucket:
                 ubucket = ubucket.strip('/')
 

celery/backends/rpc.py

@@ -5,9 +5,9 @@ RPC-style result backend, using reply-to and one queue per client.
 """
 from __future__ import absolute_import, unicode_literals
 
+import kombu
 import time
 
-from kombu import Consumer, Exchange, Producer, Queue
 from kombu.common import maybe_declare
 from kombu.utils.compat import register_after_fork
 from kombu.utils.objects import cached_property
@@ -29,7 +29,7 @@ class BacklogLimitExceeded(Exception):
     """Too much state history to fast-forward."""
 
 
-class NoCacheQueue(Queue):
+class NoCacheQueue(kombu.Queue):
     can_cache_declaration = False
 
 
@@ -38,7 +38,7 @@ def _on_after_fork_cleanup_backend(backend):
 
 
 class ResultConsumer(BaseResultConsumer):
-    Consumer = Consumer
+    Consumer = kombu.Consumer
 
     _connection = None
     _consumer = None
@@ -90,10 +90,10 @@ class ResultConsumer(BaseResultConsumer):
 class BaseRPCBackend(base.Backend, AsyncBackendMixin):
     """Base class for the RPC result backend."""
 
-    Exchange = Exchange
+    Exchange = kombu.Exchange
     Queue = NoCacheQueue
-    Consumer = Consumer
-    Producer = Producer
+    Consumer = kombu.Consumer
+    Producer = kombu.Producer
     ResultConsumer = ResultConsumer
 
     BacklogLimitExceeded = BacklogLimitExceeded
@@ -209,7 +209,7 @@ class BaseRPCBackend(base.Backend, AsyncBackendMixin):
             binding = self._create_binding(task_id)(channel)
             binding.declare()
 
-            for i in range(limit):
+            for _ in range(limit):
                 msg = binding.get(accept=accept, no_ack=no_ack)
                 if not msg:
                     break
@@ -268,14 +268,14 @@ class RPCBackend(BaseRPCBackend):
 
     persistent = False
 
-    class Consumer(Consumer):
+    class Consumer(kombu.Consumer):
         """Consumer that requires manual declaration of queues."""
 
         auto_declare = False
 
     def _create_exchange(self, name, type='direct', delivery_mode=2):
         # uses direct to queue routing (anon exchange).
-        return Exchange(None)
+        return self.Exchange(None)
 
     def _create_binding(self, task_id):
         return self.binding

celery/bin/base.py

@@ -251,7 +251,6 @@ class Command(object):
         pool_option = self.with_pool_option(argv)
         if pool_option:
             maybe_patch_concurrency(argv, *pool_option)
-            short_opts, long_opts = pool_option
 
     def usage(self, command):
         return '%prog {0} [options] {self.args}'.format(command, self=self)

celery/bin/graph.py

@@ -114,16 +114,24 @@ class graph(Command):
             pass
 
         class Backend(Node):
-            scheme = {'shape': 'folder', 'width': 2,
-                      'height': 1, 'color': 'black',
-                      'fillcolor': 'peachpuff3', 'color': 'peachpuff4'}
+            scheme = {
+                'shape': 'folder',
+                'width': 2,
+                'height': 1,
+                'color': 'black',
+                'fillcolor': 'peachpuff3',
+            }
 
             def label(self):
                 return generic_label(self) if generic else self._label
 
         class Broker(Node):
-            scheme = {'shape': 'circle', 'fillcolor': 'cadetblue3',
-                      'color': 'cadetblue4', 'height': 1}
+            scheme = {
+                'shape': 'circle',
+                'fillcolor': 'cadetblue3',
+                'color': 'cadetblue4',
+                'height': 1,
+            }
 
             def label(self):
                 return generic_label(self) if generic else self._label

celery/concurrency/asynpool.py

@@ -35,7 +35,7 @@ from billiard.pool import RUN, TERMINATE, ACK, NACK, WorkersJoined
 from billiard import pool as _pool
 from billiard.compat import buf_t, setblocking, isblocking
 from billiard.queues import _SimpleQueue
-from kombu.async import READ, WRITE, ERR
+from kombu.async import WRITE, ERR
 from kombu.serialization import pickle as _pickle
 from kombu.utils.eventio import SELECT_BAD_FD
 from kombu.utils.functional import fxrange
@@ -506,7 +506,7 @@ class AsynPool(_pool.Pool):
             _discard_tref(R._job)
         self.on_timeout_cancel = on_timeout_cancel
 
-    def _on_soft_timeout(self, job, soft, hard, hub):
+    def _on_soft_timeout(self, job, soft, hard, hub, now=time.time):
         # only used by async pool.
         if hard:
             self._tref_for_id[job] = hub.call_at(

celery/worker/consumer/connection.py

@@ -29,7 +29,8 @@ class Connection(bootsteps.StartStopStep):
         if connection:
             ignore_errors(connection, connection.close)
 
-    def info(self, c, params='N/A'):
+    def info(self, c):
+        params = 'N/A'
         if c.connection:
             params = c.connection.info()
             params.pop('password', None)  # don't send password.

celery/worker/request.py

@@ -490,8 +490,7 @@ class Request(object):
 
 def create_request_cls(base, task, pool, hostname, eventer,
                        ref=ref, revoked_tasks=revoked_tasks,
-                       task_ready=task_ready):
-    from celery.app.trace import trace_task_ret as trace
+                       task_ready=task_ready, trace=trace_task_ret):
     default_time_limit = task.time_limit
     default_soft_time_limit = task.soft_time_limit
     apply_async = pool.apply_async

examples/next-steps/setup.py

@@ -11,12 +11,30 @@ from setuptools import setup, find_packages
 
 setup(
     name='example-tasks',
+    url='http://github.com/example/celery-tasks',
+    author='Ola A. Normann',
+    author_email='author@example.com',
+    keywords='our celery integration',
     version='1.0',
     description='Tasks for my project',
+    long_description=__doc__,
+    license='BSD',
     packages=find_packages(exclude=['ez_setup', 'tests', 'tests.*']),
+    test_suite='nose.collector',
     zip_safe=False,
     install_requires=[
         'celery>=4.0',
         #  'requests',
     ],
+    classifiers=[
+        'Development Status :: 5 - Production/Stable',
+        'License :: OSI Approved :: BSD License',
+        'Programming Language :: Python :: 2',
+        'Programming Language :: Python :: 2.7',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.5',
+        'Programming Language :: Python :: Implementation :: CPython',
+        'Programming Language :: Python :: Implementation :: PyPy',
+        'Operating System :: OS Independent',
+    ],
 )

t/integration/setup.py

@@ -44,6 +44,8 @@ setup(
     author='Ask Solem',
     author_email='ask@celeryproject.org',
     url='https://github.com/celery/celery',
+    keywords='celery integration tests',
+    license='BSD',
     platforms=['any'],
     packages=[],
     data_files=[],
@@ -62,6 +64,14 @@ setup(
         'Programming Language :: Python',
         'License :: OSI Approved :: BSD License',
         'Intended Audience :: Developers',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 2',
+        'Programming Language :: Python :: 2.7',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.4',
+        'Programming Language :: Python :: 3.5',
+        'Programming Language :: Python :: Implementation :: CPython',
+        'Programming Language :: Python :: Implementation :: PyPy',
     ],
-    long_description='Do not install this package',
+    long_description=__doc__,
 )