Flake8

celery/__compat__.py

@@ -80,6 +80,7 @@ COMPAT_MODULES = {
             "inspect": "control.inspect",
         },
         "schedules": "celery.schedules",
+        "chords": "celery.canvas",
     }
 }
 

celery/app/amqp.py

@@ -15,12 +15,12 @@ from datetime import timedelta
 from weakref import WeakValueDictionary
 
 from kombu import BrokerConnection, Consumer, Exchange, Producer, Queue
-from kombu.common import entry_to_queue, maybe_declare
+from kombu.common import entry_to_queue
 from kombu.pools import ProducerPool
 
 from celery import signals
-from celery.utils import cached_property, lpmerge, uuid
-from celery.utils import text
+from celery.utils import cached_property, uuid
+from celery.utils.text import indent as textindent
 
 from . import routes as _routes
 
@@ -109,8 +109,8 @@ class Queues(dict):
                     "routing_key": q.routing_key}
                         for name, q in sorted(active.iteritems())]
         if indent_first:
-            return text.indent("\n".join(info), indent)
-        return info[0] + "\n" + text.indent("\n".join(info[1:]), indent)
+            return textindent("\n".join(info), indent)
+        return info[0] + "\n" + textindent("\n".join(info[1:]), indent)
 
     def select_subset(self, wanted):
         """Sets :attr:`consume_from` by selecting a subset of the
@@ -156,11 +156,9 @@ class TaskProducer(Producer):
             routing_key=None, serializer=None, delivery_mode=None,
             compression=None, **kwargs):
         """Send task message."""
-
-        connection = self.connection
-        _retry_policy = self.retry_policy
-        if retry_policy:  # merge default and custom policy
-            _retry_policy = dict(_retry_policy, **retry_policy)
+        # merge default and custom policy
+        _rp = (dict(self.retry_policy, **retry_policy) if retry_policy
+                                                       else self.retry_policy)
         task_id = task_id or uuid()
         task_args = task_args or []
         task_kwargs = task_kwargs or {}
@@ -168,7 +166,7 @@ class TaskProducer(Producer):
             raise ValueError("task args must be a list or tuple")
         if not isinstance(task_kwargs, dict):
             raise ValueError("task kwargs must be a dictionary")
-        if countdown:                           # Convert countdown to ETA.
+        if countdown:  # Convert countdown to ETA.
             now = now or self.app.now()
             eta = now + timedelta(seconds=countdown)
         if isinstance(expires, (int, float)):
@@ -179,8 +177,8 @@ class TaskProducer(Producer):
 
         body = {"task": task_name,
                 "id": task_id,
-                "args": task_args or [],
-                "kwargs": task_kwargs or {},
+                "args": task_args,
+                "kwargs": task_kwargs,
                 "retries": retries or 0,
                 "eta": eta,
                 "expires": expires,
@@ -195,9 +193,8 @@ class TaskProducer(Producer):
         self.publish(body, exchange=exchange, mandatory=mandatory,
              immediate=immediate, routing_key=routing_key,
              serializer=serializer or self.serializer,
-             delivery_mode=delivery_mode,
              compression=compression or self.compression,
-             retry=retry, retry_policy=retry_policy,
+             retry=retry, retry_policy=_rp, delivery_mode=delivery_mode,
              declare=[self.app.amqp.queues[queue]] if queue else [])
 
         signals.task_sent.send(sender=task_name, **body)
@@ -298,8 +295,8 @@ class AMQP(object):
         """Queue name⇒ declaration mapping."""
         return self.Queues(self.app.conf.CELERY_QUEUES)
 
-    @queues.setter
-    def queues(self, queues):  # noqa
+    @queues.setter  # noqa
+    def queues(self, queues):
         return self.Queues(queues)
 
     @property

celery/app/base.py

@@ -209,7 +209,6 @@ class Celery(object):
             with self.amqp.publisher_pool.acquire(block=True) as producer:
                 yield producer
 
-
     def with_default_connection(self, fun):
         """With any function accepting a `connection`
         keyword argument, establishes a default connection if one is

celery/app/builtins.py

@@ -81,7 +81,6 @@ def add_map_task(app):
         return list(map(task, it))
 
 
-
 @builtin_task
 def add_starmap_task(app):
     from celery.canvas import subtask
@@ -144,7 +143,8 @@ def add_group_task(app):
             if self.app.conf.CELERY_ALWAYS_EAGER:
                 return self.apply(args, kwargs, **options)
             tasks, result, gid = self.prepare(options, **kwargs)
-            super(Group, self).apply_async((list(tasks), result, gid), **options)
+            super(Group, self).apply_async(
+                    (list(tasks), result, gid), **options)
             return result
 
         def apply(self, args=(), kwargs={}, **options):

celery/task/chords.py

@@ -1,2 +0,0 @@
-# compat module
-from celery.canvas import Chord, chord  # noqa

celery/task/schedules.py

@@ -1,9 +0,0 @@
-# -*- coding: utf-8 -*-
-from __future__ import absolute_import
-
-import warnings
-from celery.schedules import schedule, crontab_parser, crontab  # noqa
-from celery.exceptions import CDeprecationWarning
-
-warnings.warn(CDeprecationWarning(
-    "celery.task.schedules is deprecated and renamed to celery.schedules"))

celery/task/sets.py

@@ -2,6 +2,7 @@
 from __future__ import absolute_import
 from __future__ import with_statement
 
+from celery.app import app_or_default
 from celery.app.state import get_current_task
 from celery.canvas import subtask, maybe_subtask  # noqa
 from celery.utils import uuid
@@ -22,10 +23,10 @@ class TaskSet(UserList):
         >>> list_of_return_values = taskset_result.join()  # *expensive*
 
     """
-    _app = None
+    app = None
 
     def __init__(self, tasks=None, app=None, Publisher=None):
-        self.app = app_or_default(app)
+        self.app = app_or_default(app or self.app)
         self.data = [maybe_subtask(t) for t in tasks or []]
         self.total = len(self.tasks)
         self.Publisher = Publisher or self.app.amqp.TaskProducer

celery/tests/app/test_app.py

@@ -263,8 +263,6 @@ class test_App(Case):
             chan.close()
         assert conn.transport_cls == "memory"
 
-        entities = conn.declared_entities
-
         pub = self.app.amqp.TaskPublisher(conn,
                 exchange=Exchange("foo_exchange"))
 

celery/tests/app/test_routes.py

@@ -3,7 +3,6 @@ from __future__ import with_statement
 
 from functools import wraps
 
-from kombu import Exchange
 from kombu.utils.functional import maybe_promise
 
 from celery import current_app

celery/tests/tasks/test_sets.py

@@ -3,8 +3,6 @@ from __future__ import with_statement
 
 import anyjson
 
-from mock import Mock
-
 from celery import current_app
 from celery.task import Task
 from celery.task.sets import subtask, TaskSet
@@ -175,7 +173,7 @@ class test_TaskSet(Case):
     def test_set_app(self):
         ts = TaskSet([])
         ts.app = 42
-        self.assertEqual(ts._app, 42)
+        self.assertEqual(ts.app, 42)
 
     def test_set_tasks(self):
         ts = TaskSet([])
@@ -185,4 +183,4 @@ class test_TaskSet(Case):
     def test_set_Publisher(self):
         ts = TaskSet([])
         ts.Publisher = 42
-        self.assertEqual(ts._Publisher, 42)
+        self.assertEqual(ts.Publisher, 42)

contrib/release/doc4allmods

@@ -10,7 +10,6 @@ SKIP_FILES="celery.__compat__.rst
             celery.concurrency.processes._win.rst
             celery.contrib.rst
             celery.contrib.bundles.rst
-            celery.task.schedules.rst
             celery.local.rst
             celery.app.base.rst
             celery.apps.rst