flakes

.landscape.yml

@@ -41,5 +41,11 @@ pylint:
         - too-many-instance-attributes
         - bad-builtin
         - abstract-method
+        - global-statement
+        - too-many-public-methods
+        - no-self-use
+        - unnecessary-lambda
+        - too-few-public-methods
+        - attribute-defined-outside-init
     options:
         exclude-protected: _reader, _writer, _popen, _sentinel_poll, _job, _is_alive, _write_to, _scheduled_for, _terminated, _accepted, _set_terminated, _payload, _cancel

celery/__init__.py

@@ -21,6 +21,7 @@
 __contact__ = '[email protected]'
 __homepage__ = 'http://celeryproject.org'
 __docformat__ = 'restructuredtext'
+__keywords__ = 'task job queue distributed messaging actor'
 
 # -eof meta-
 

celery/__main__.py

@@ -12,8 +12,8 @@ def main():
     """Entrypoint to the ``celery`` umbrella command."""
     if 'multi' not in sys.argv:
         maybe_patch_concurrency()
-    from celery.bin.celery import main
-    main()
+    from celery.bin.celery import main as _main
+    _main()
 
 
 if __name__ == '__main__':  # pragma: no cover

celery/_state.py

@@ -90,7 +90,9 @@ def _set_current_app(app):
 if os.environ.get('C_STRICT_APP'):  # pragma: no cover
     def get_current_app():
         """Return the current app."""
-        raise Exception('USES CURRENT APP')
+        raise RuntimeError('USES CURRENT APP')
+elif os.environ.get('C_WARN_APP'):  # pragma: no cover
+    def get_current_app():  # noqa
         import traceback
         print('-- USES CURRENT_APP', file=sys.stderr)  # noqa+
         traceback.print_stack(file=sys.stderr)

celery/app/trace.py

@@ -12,6 +12,11 @@
 # but in the end it only resulted in bad performance and horrible tracebacks,
 # so instead we now use one closure per task class.
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+# pylint: disable=broad-except
+# We know what we're doing...
+
 import logging
 import os
 import sys
@@ -181,7 +186,7 @@ def handle_retry(self, task, req, store_errors=True, **kwargs):
             })
             return einfo
         finally:
-            del(tb)
+            del tb
 
     def handle_failure(self, task, req, store_errors=True, call_errbacks=True):
         """Handle exception."""
@@ -208,7 +213,7 @@ def handle_failure(self, task, req, store_errors=True, call_errbacks=True):
             self._log_error(task, req, einfo)
             return einfo
         finally:
-            del(tb)
+            del tb
 
     def _log_error(self, task, req, einfo):
         eobj = einfo.exception = get_pickled_exception(einfo.exception)
@@ -535,7 +540,7 @@ def report_internal_error(task, exc):
                 exc, exc_info.traceback)))
         return exc_info
     finally:
-        del(_tb)
+        del _tb
 
 
 def setup_worker_optimizations(app, hostname=None):

celery/bin/celery.py

@@ -346,8 +346,8 @@ def get_options(self):
 
     def run_from_argv(self, prog_name, argv, command=None):
         from celery.bin.multi import MultiTool
-        multi = MultiTool(quiet=self.quiet, no_color=self.no_color)
-        return multi.execute_from_commandline([command] + argv)
+        cmd = MultiTool(quiet=self.quiet, no_color=self.no_color)
+        return cmd.execute_from_commandline([command] + argv)
 
 
 class list_(Command):
@@ -535,11 +535,11 @@ def run(self, task_id, *args, **kwargs):
 
         if task:
             result_cls = self.app.tasks[task].AsyncResult
-        result = result_cls(task_id)
+        task_result = result_cls(task_id)
         if traceback:
-            value = result.traceback
+            value = task_result.traceback
         else:
-            value = result.get()
+            value = task_result.get()
         self.out(self.pretty(value)[1])
 
 
@@ -849,6 +849,8 @@ def run(self, ipython=False, bpython=False,
         import celery
         import celery.task.base
         self.app.loader.import_default_modules()
+
+        # pylint: disable=attribute-defined-outside-init
         self.locals = {
             'app': self.app,
             'celery': self.app,
@@ -1136,7 +1138,7 @@ def prepare_prog_name(self, name):
             return sys.modules['__main__'].__file__
         return name
 
-    def handle_argv(self, prog_name, argv):
+    def handle_argv(self, prog_name, argv, **kwargs):
         self.prog_name = self.prepare_prog_name(prog_name)
         argv = self._relocate_args_from_start(argv)
         _, argv = self.prepare_args(None, argv)
@@ -1204,12 +1206,5 @@ def load_extension_commands(self):
             command_classes.append(('Extensions', names, 'magenta'))
 
 
-def command(*args, **kwargs):
-    # Deprecated: Use classmethod
-    #             :meth:`CeleryCommand.register_command` instead.
-    _register = CeleryCommand.register_command
-    return _register(args[0]) if args else _register
-
-
 if __name__ == '__main__':          # pragma: no cover
     main()

celery/bin/graph.py

@@ -35,12 +35,12 @@ def bootsteps(self, *args, **kwargs):
         worker = self.app.WorkController()
         include = {arg.lower() for arg in args or ['worker', 'consumer']}
         if 'worker' in include:
-            graph = worker.blueprint.graph
+            worker_graph = worker.blueprint.graph
             if 'consumer' in include:
                 worker.blueprint.connect_with(worker.consumer.blueprint)
         else:
-            graph = worker.consumer.blueprint.graph
-        graph.to_dot(self.stdout)
+            worker_graph = worker.consumer.blueprint.graph
+        worker_graph.to_dot(self.stdout)
 
     def workers(self, *args, **kwargs):
 
@@ -73,14 +73,21 @@ def __str__(self):
                 return self.label()
 
         class Thread(Node):
-            scheme = {'fillcolor': 'lightcyan4', 'fontcolor': 'yellow',
-                      'shape': 'oval', 'fontsize': 10, 'width': 0.3,
-                      'color': 'black'}
+            scheme = {
+                'fillcolor': 'lightcyan4',
+                'fontcolor': 'yellow',
+                'shape': 'oval',
+                'fontsize': 10,
+                'width': 0.3,
+                'color': 'black',
+            }
 
             def __init__(self, label, **kwargs):
-                self._label = 'thr-{0}'.format(next(tids))
                 self.real_label = label
-                self.pos = 0
+                super(Thread, self).__init__(
+                    label='thr-{0}'.format(next(tids)),
+                    pos=0,
+                )
 
         class Formatter(GraphFormatter):
 
@@ -177,24 +184,24 @@ def maybe_abbr(l, name, max=Wmax):
         broker = Broker(args.get(
             'broker', self.app.connection_for_read().as_uri()))
         backend = Backend(backend) if backend else None
-        graph = DependencyGraph(formatter=Formatter())
-        graph.add_arc(broker)
+        deps = DependencyGraph(formatter=Formatter())
+        deps.add_arc(broker)
         if backend:
-            graph.add_arc(backend)
+            deps.add_arc(backend)
         curworker = [0]
         for i, worker in enumerate(workers):
             worker = Worker(worker, pos=i)
-            graph.add_arc(worker)
-            graph.add_edge(worker, broker)
+            deps.add_arc(worker)
+            deps.add_edge(worker, broker)
             if backend:
-                graph.add_edge(worker, backend)
+                deps.add_edge(worker, backend)
             threads = threads_for.get(worker._label)
             if threads:
                 for thread in threads:
                     thread = Thread(thread)
-                    graph.add_arc(thread)
-                    graph.add_edge(thread, worker)
+                    deps.add_arc(thread)
+                    deps.add_edge(thread, worker)
 
             curworker[0] += 1
 
-        graph.to_dot(self.stdout)
+        deps.to_dot(self.stdout)

celery/bootsteps.py

@@ -151,7 +151,7 @@ def send_all(self, parent, method,
                                 description.capitalize(), step.alias)
                     try:
                         fun(parent, *args)
-                    except Exception as exc:
+                    except Exception as exc:  # pylint: ignore=broad-except
                         if propagate:
                             raise
                         logger.exception(
@@ -274,6 +274,9 @@ def alias(self):
 class StepType(type):
     """Meta-class for steps."""
 
+    name = None
+    requires = None
+
     def __new__(cls, name, bases, attrs):
         module = attrs.get('__module__')
         qname = '{0}.{1}'.format(module, name) if module else name

celery/concurrency/asynpool.py

@@ -46,6 +46,9 @@
 from celery.utils.log import get_logger
 from celery.worker import state as worker_state
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+
 try:
     from _billiard import read as __read__
     from struct import unpack_from as _unpack_from

celery/concurrency/gevent.py

@@ -2,7 +2,6 @@
 """Gevent execution pool."""
 from __future__ import absolute_import, unicode_literals
 
-
 try:
     from gevent import Timeout
 except ImportError:  # pragma: no cover
@@ -15,6 +14,9 @@
 
 __all__ = ['TaskPool']
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+
 
 def apply_timeout(target, args=(), kwargs={}, callback=None,
                   accept_callback=None, pid=None, timeout=None,
@@ -101,7 +103,7 @@ def on_stop(self):
 
     def on_apply(self, target, args=None, kwargs=None, callback=None,
                  accept_callback=None, timeout=None,
-                 timeout_callback=None, **_):
+                 timeout_callback=None, apply_target=base.apply_target, **_):
         timeout = self.timeout if timeout is None else timeout
         return self._quick_put(apply_timeout if timeout else apply_target,
                                target, args, kwargs, callback, accept_callback,

celery/events/__init__.py

@@ -30,6 +30,9 @@
 
 __all__ = ['Events', 'Event', 'EventDispatcher', 'EventReceiver']
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+
 event_exchange = Exchange('celeryev', type='topic')
 
 _TZGETTER = itemgetter('utcoffset', 'timestamp')
@@ -208,7 +211,7 @@ def _publish(self, event, producer, routing_key, retry=False,
                 headers=self.headers,
                 delivery_mode=self.delivery_mode,
             )
-        except Exception as exc:
+        except Exception as exc:  # pylint: disable=broad-except
             if not self.buffer_while_offline:
                 raise
             self._outbound_buffer.append((event, routing_key, exc))
@@ -430,6 +433,8 @@ def State(self):
     def default_dispatcher(self, hostname=None, enabled=True,
                            buffer_while_offline=False):
         with self.app.amqp.producer_pool.acquire(block=True) as prod:
+            # pylint: disable=too-many-function-args
+            # This is a property pylint...
             with self.Dispatcher(prod.connection, hostname, enabled,
                                  prod.channel, buffer_while_offline) as d:
                 yield d

celery/events/state.py

@@ -37,15 +37,21 @@
 
 __all__ = ['Worker', 'Task', 'State', 'heartbeat_expires']
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+# pylint: disable=too-many-function-args
+# For some reason pylint thinks ._event is a method, when it's a property.
+
+#: Set if running PyPy
 PYPY = hasattr(sys, 'pypy_version_info')
 
-# The window (in percentage) is added to the workers heartbeat
-# frequency.  If the time between updates exceeds this window,
-# then the worker is considered to be offline.
+#: The window (in percentage) is added to the workers heartbeat
+#: frequency.  If the time between updates exceeds this window,
+#: then the worker is considered to be offline.
 HEARTBEAT_EXPIRE_WINDOW = 200
 
-# Max drift between event timestamp and time of event received
-# before we alert that clocks may be unsynchronized.
+#: Max drift between event timestamp and time of event received
+#: before we alert that clocks may be unsynchronized.
 HEARTBEAT_DRIFT_MAX = 16
 
 DRIFT_WARNING = """\

celery/utils/pytest.py

@@ -19,6 +19,10 @@
 from celery.app import current_app
 from celery.backends.cache import CacheBackend, DummyClient
 
+# pylint: disable=redefined-outer-name
+# Well, they're called fixtures....
+
+
 CELERY_TEST_CONFIG = {
     #: Don't want log output when running suite.
     'worker_hijack_root_logger': False,
@@ -68,12 +72,12 @@ def __init__(self, *args, **kwargs):
 def TestApp(name=None, set_as_current=False, log=UnitLogging,
             broker='memory://', backend='cache+memory://', **kwargs):
     """App used for testing."""
-    app = Celery(name or 'celery.tests',
-                 set_as_current=set_as_current,
-                 log=log, broker=broker, backend=backend,
-                 **kwargs)
-    app.add_defaults(deepcopy(CELERY_TEST_CONFIG))
-    return app
+    test_app = Celery(
+        name or 'celery.tests',
+        set_as_current=set_as_current,
+        log=log, broker=broker, backend=backend, **kwargs)
+    test_app.add_defaults(deepcopy(CELERY_TEST_CONFIG))
+    return test_app
 
 
 @pytest.fixture(autouse=True)
@@ -92,22 +96,22 @@ class NonTLS(object):
         current_app = trap
     _state._tls = NonTLS()
 
-    app = TestApp(set_as_current=False)
+    test_app = TestApp(set_as_current=False)
     is_not_contained = any([
         not getattr(request.module, 'app_contained', True),
         not getattr(request.cls, 'app_contained', True),
         not getattr(request.function, 'app_contained', True)
     ])
     if is_not_contained:
-        app.set_current()
+        test_app.set_current()
 
-    yield app
+    yield test_app
 
     _state.set_default_app(prev_default_app)
     _state._tls = prev_tls
     _state._tls.current_app = prev_current_app
-    if app is not prev_current_app:
-        app.close()
+    if test_app is not prev_current_app:
+        test_app.close()
     _state._on_app_finalizers = prev_finalizers
     _state._apps = prev_apps
 

celery/utils/saferepr.py

@@ -29,6 +29,9 @@
 
 __all__ = ['saferepr', 'reprstream']
 
+# pylint: disable=redefined-outer-name
+# We cache globals and attribute lookups, so disable this warning.
+
 IS_PY3 = sys.version_info[0] == 3
 
 if IS_PY3:  # pragma: no cover