PEP8ify + pyflakes

celery/concurrency/processes/__init__.py

@@ -8,7 +8,7 @@ from celery import log
 from celery.datastructures import ExceptionInfo
 from celery.utils.functional import curry
 
-from celery.concurrency.processes.pool import Pool, RUN, MaybeEncodingError
+from celery.concurrency.processes.pool import Pool, RUN
 
 
 class TaskPool(object):

celery/concurrency/processes/pool.py

@@ -51,10 +51,6 @@ def mapstar(args):
 #
 
 
-def soft_timeout_sighandler(signum, frame):
-    raise SoftTimeLimitExceeded()
-
-
 class MaybeEncodingError(Exception):
     """Wraps unpickleable object."""
 
@@ -63,12 +59,17 @@ class MaybeEncodingError(Exception):
         self.value = repr(value)
         super(MaybeEncodingError, self).__init__(self.exc, self.value)
 
-    def __str__(self):
-        return "Error sending result: '%s'. Reason: '%s'." % (self.value,
-                                                              self.exc)
     def __repr__(self):
         return "<MaybeEncodingError: %s>" % str(self)
 
+    def __str__(self):
+        return "Error sending result: '%s'. Reason: '%s'." % (
+                    self.value, self.exc)
+
+
+def soft_timeout_sighandler(signum, frame):
+    raise SoftTimeLimitExceeded()
+
 
 def worker(inqueue, outqueue, ackqueue, initializer=None, initargs=(),
         maxtasks=None):
@@ -109,7 +110,6 @@ def worker(inqueue, outqueue, ackqueue, initializer=None, initargs=(),
             put((job, i, result))
         except Exception, exc:
             wrapped = MaybeEncodingError(exc, result[1])
-            debug('Got possible encoding error while sending result: %s' % wrapped)
             put((job, i, (False, wrapped)))
 
         completed += 1

celery/db/dfd042c7.py

@@ -1,6 +1,9 @@
 """
 dfd042c7
-SQLAlchemy 0.5.8 version of a805d4bd, see the docstring of that module for an explanation of this workaround.
+
+SQLAlchemy 0.5.8 version of a805d4bd, see the docstring of that module
+for an explanation of this workaround.
+
 """
 from sqlalchemy.types import PickleType as _PickleType
 from sqlalchemy import util
@@ -31,11 +34,14 @@ class PickleType(_PickleType):
         if self.comparator:
             return self.comparator(x, y)
         elif self.mutable and not hasattr(x, '__eq__') and x is not None:
-            util.warn_deprecated("Objects stored with PickleType when mutable=True must implement __eq__() for reliable comparison.")
-            return self.pickler.dumps(x, self.protocol) == self.pickler.dumps(y, self.protocol)
+            util.warn_deprecated(
+                    "Objects stored with PickleType when mutable=True "
+                    "must implement __eq__() for reliable comparison.")
+            a = self.pickler.dumps(x, self.protocol)
+            b = self.pickler.dumps(y, self.protocol)
+            return a == b
         else:
             return x == y
 
     def is_mutable(self):
         return self.mutable
-

celery/tests/test_task_sets.py

@@ -102,7 +102,7 @@ class test_TaskSet(unittest.TestCase):
             return ts
 
         context = catch_warnings(record=True)
-        ts = execute_context(context, with_catch_warnings)
+        execute_context(context, with_catch_warnings)
 
         # TaskSet.task (deprecated)
         def with_catch_warnings2(log):
@@ -111,7 +111,6 @@ class test_TaskSet(unittest.TestCase):
             self.assertTrue(log)
             self.assertIn("TaskSet.task is deprecated",
                           log[0].message.args[0])
-            return ts
 
         execute_context(catch_warnings(record=True), with_catch_warnings2)
 

celery/tests/test_utils.py

@@ -101,7 +101,6 @@ class test_utils(unittest.TestCase):
         self.assertEqual("four", utils.firstmethod("m")([
             A(), A(), A(), promise(lambda: A("four")), A("five")]))
 
-
     def test_first(self):
         iterations = [0]
 

celery/utils/__init__.py

@@ -88,6 +88,7 @@ def maybe_promise(value):
         return value.evaluate()
     return value
 
+
 def noop(*args, **kwargs):
     """No operation.
 
@@ -341,5 +342,3 @@ def instantiate(name, *args, **kwargs):
 
     """
     return get_cls_by_name(name)(*args, **kwargs)
-
-

celery/utils/compat.py

@@ -366,7 +366,7 @@ def _compat_chain_from_iterable(iterables):
         for element in it:
             yield element
 
-#try:
-#    chain_from_iterable = getattr(chain, "from_iterable")
-#except AttributeError:
-chain_from_iterable = _compat_chain_from_iterable
+try:
+    chain_from_iterable = getattr(chain, "from_iterable")
+except AttributeError:
+    chain_from_iterable = _compat_chain_from_iterable

celery/worker/control/builtins.py

@@ -3,7 +3,6 @@ from datetime import datetime
 from celery import conf
 from celery.backends import default_backend
 from celery.registry import tasks
-from celery.serialization import pickle
 from celery.utils import timeutils
 from celery.worker import state
 from celery.worker.state import revoked

celery/worker/job.py

@@ -403,15 +403,13 @@ class TaskRequest(object):
                                        exception=repr(exc_info.exception),
                                        traceback=exc_info.traceback)
 
-        context = {
-            "hostname": self.hostname,
-            "id": self.task_id,
-            "name": self.task_name,
-            "exc": repr(exc_info.exception),
-            "traceback": unicode(exc_info.traceback, 'utf-8'),
-            "args": self.args,
-            "kwargs": self.kwargs,
-        }
+        context = {"hostname": self.hostname,
+                   "id": self.task_id,
+                   "name": self.task_name,
+                   "exc": repr(exc_info.exception),
+                   "traceback": unicode(exc_info.traceback, 'utf-8'),
+                   "args": self.args,
+                   "kwargs": self.kwargs}
         self.logger.error(self.error_msg.strip() % context)
 
         task_obj = tasks.get(self.task_name, object)

contrib/release/doc4allmods

@@ -7,7 +7,8 @@ SKIP_FILES="celery.bin.rst celery.contrib.rst
             celery.models.rst
             celery.concurrency.rst
             celery.db.rst
-            celery.db.a805d4bd.rst"
+            celery.db.a805d4bd.rst
+            celery.db.dfd042c7.rst"
 
 modules=$(find "$PACKAGE" -name "*.py")
 

docs/conf.py

@@ -65,7 +65,7 @@ latex_documents = [
 html_theme = "celery"
 html_theme_path = ["_theme"]
 html_sidebars = {
-    'index':    ['sidebarintro.html', 'sourcelink.html', 'searchbox.html'],
-    '**':       ['sidebarlogo.html', 'localtoc.html', 'relations.html',
-                 'sourcelink.html', 'searchbox.html']
+    'index': ['sidebarintro.html', 'sourcelink.html', 'searchbox.html'],
+    '**': ['sidebarlogo.html', 'localtoc.html', 'relations.html',
+           'sourcelink.html', 'searchbox.html'],
 }

docs/internals/reference/celery.worker.revoke.rst

@@ -1,10 +0,0 @@
-==============================================
- Worker Revoked Tasks - celery.worker.revoke
-==============================================
-
-.. data:: revoked
-
-    A :class:`celery.datastructures.LimitedSet` containing revoked task ids.
-
-    Items expire after one hour, and the structure can only hold
-    10000 expired items at a time (about 300kb).

docs/internals/reference/celery.worker.state.rst

@@ -0,0 +1,11 @@
+====================================
+ Worker State - celery.worker.state
+====================================
+
+.. contents::
+    :local:
+.. currentmodule:: celery.worker.state
+
+.. automodule:: celery.worker.state
+    :members:
+    :undoc-members:

docs/internals/reference/index.rst

@@ -18,7 +18,7 @@
     celery.worker.control
     celery.worker.control.builtins
     celery.worker.control.registry
-    celery.worker.revoke
+    celery.worker.state
     celery.concurrency.processes
     celery.concurrency.processes.pool
     celery.concurrency.threads