PEP8ify + pyflakes

celery/app/base.py

@@ -8,7 +8,6 @@ Application Base Class.
 :license: BSD, see LICENSE for more details.
 
 """
-import os
 import platform as _platform
 
 from copy import deepcopy
@@ -248,7 +247,6 @@ class BaseApp(object):
         @wraps(fun)
         def _inner(*args, **kwargs):
             connection = kwargs.get("connection")
-            timeout = kwargs.get("connect_timeout")
             kwargs["connection"] = conn = connection or \
                     self.pool.acquire(block=True)
             close_connection = not connection and conn.release or None
@@ -320,7 +318,8 @@ class BaseApp(object):
                 register_after_fork(self, self._after_fork)
             except ImportError:
                 pass
-            self._pool = self.broker_connection().Pool(self.conf.BROKER_POOL_LIMIT)
+            self._pool = self.broker_connection().Pool(
+                            self.conf.BROKER_POOL_LIMIT)
         return self._pool
 
     @cached_property

celery/result.py

@@ -435,7 +435,8 @@ class TaskSetResult(ResultSet):
     """An instance of this class is returned by
     `TaskSet`'s :meth:`~celery.task.TaskSet.apply_async` method.
 
-    It enables inspection of the tasks state and return values as a single entity.
+    It enables inspection of the tasks state and return values as
+    a single entity.
 
     :param taskset_id: The id of the taskset.
     :param results: List of result instances.

celery/task/base.py

@@ -433,8 +433,6 @@ class BaseTask(object):
                            conf.CELERY_MESSAGE_COMPRESSION)
         options = dict(extract_exec_options(self), **options)
         options = router.route(options, self.name, args, kwargs)
-        exchange = options.get("exchange")
-        exchange_type = options.get("exchange_type")
         expires = expires or self.expires
 
         publish = publisher or self.app.amqp.publisher_pool.acquire(block=True)

celery/task/chord.py

@@ -4,6 +4,7 @@ from celery import current_app
 from celery.result import TaskSetResult
 from celery.task.sets import TaskSet, subtask
 
+
 @current_app.task(name="celery.chord_unlock", max_retries=None)
 def _unlock_chord(setid, callback, interval=1, max_retries=None):
     result = TaskSetResult.restore(setid)
@@ -24,7 +25,7 @@ class Chord(current_app.Task):
             uuid = gen_unique_id()
             task.options.update(task_id=uuid, chord=body)
             r.append(current_app.AsyncResult(uuid))
-        ts = current_app.TaskSetResult(setid, r).save()
+        current_app.TaskSetResult(setid, r).save()
         self.backend.on_chord_apply(setid, body)
         return set.apply_async(taskset_id=setid)
 

contrib/release/flakesignore.txt

@@ -67,10 +67,12 @@ celery/tests/test_concurrency/test_concurrency_processes.py:103: 'multiprocessin
 celery/tests/test_events/test_events_cursesmon.py:16: 'curses' imported but unused
 celery/tests/test_task/test_task_http.py:9: redefinition of unused 'addinfourl' from line 7
 celery/tests/test_task/test_task_http.py:13: redefinition of unused 'contextmanager' from line 11
-celery/backends/pyredis.py:90: redefinition of function 'client' from line 83
 celery/utils/__init__.py:16: 'cached_property' imported but unused
 celery/beat.py:16: redefinition of unused 'multiprocessing' from line 14
 celery/worker/state.py:62: redefinition of function 'task_reserved' from line 36
 celery/worker/state.py:68: redefinition of function 'task_ready' from line 47
 celery/beat.py:11: redefinition of unused 'multiprocessing' from line 9
 celery/utils/compat.py:498: redefinition of unused 'WatchedFileHandler' from line 496
+celery/backends/pyredis.py:103: redefinition of function 'client' from line 96
+celery/task/__init__.py:7: 'chord' imported but unused
+