Fix flake8 errors.

celery/__init__.py

@@ -145,6 +145,7 @@ def maybe_patch_concurrency(argv=sys.argv,
         from celery import concurrency
         concurrency.get_implementation(pool)
 
+
 # Lazy loading
 from . import local  # noqa
 

celery/app/amqp.py

@@ -520,7 +520,7 @@ class AMQP(object):
 
             # convert to anon-exchange, when exchange not set and direct ex.
             if (not exchange or not routing_key) and exchange_type == 'direct':
-                    exchange, routing_key = '', qname
+                exchange, routing_key = '', qname
             elif exchange is None:
                 # not topic exchange, and exchange not undefined
                 exchange = queue.exchange.name or default_exchange

celery/app/base.py

@@ -1243,4 +1243,6 @@ class Celery(object):
                 if not conf.timezone:
                     return timezone.local
         return timezone.get_timezone(tz)
+
+
 App = Celery  # noqa: E305 XXX compat

celery/app/task.py

@@ -1005,4 +1005,6 @@ class Task(object):
     @property
     def __name__(self):
         return self.__class__.__name__
+
+
 BaseTask = Task  # noqa: E305 XXX compat alias

celery/app/trace.py

@@ -44,9 +44,6 @@ from celery.utils.serialization import (get_pickleable_etype,
 # We know what we're doing...
 
 
-
-
-
 __all__ = [
     'TraceInfo', 'build_tracer', 'trace_task',
     'setup_worker_optimizations', 'reset_worker_optimizations',
@@ -519,6 +516,8 @@ def _trace_task_ret(name, uuid, request, body, content_type,
     R, I, T, Rstr = trace_task(app.tasks[name],
                                uuid, args, kwargs, request, app=app)
     return (1, R, T) if I else (0, Rstr, T)
+
+
 trace_task_ret = _trace_task_ret  # noqa: E305
 
 

celery/apps/beat.py

@@ -117,9 +117,9 @@ class Beat(object):
         c = self.colored
         return text_t(  # flake8: noqa
             c.blue('__    ', c.magenta('-'),
-            c.blue('    ... __   '), c.magenta('-'),
-            c.blue('        _\n'),
-            c.reset(self.startup_info(service))),
+                   c.blue('    ... __   '), c.magenta('-'),
+                   c.blue('        _\n'),
+                   c.reset(self.startup_info(service))),
         )
 
     def init_loader(self):

celery/backends/base.py

@@ -502,6 +502,8 @@ class SyncBackendMixin(object):
 
 class BaseBackend(Backend, SyncBackendMixin):
     """Base (synchronous) result backend."""
+
+
 BaseDictBackend = BaseBackend  # noqa: E305 XXX compat
 
 

celery/backends/mongodb.py

@@ -121,11 +121,11 @@ class MongoBackend(BaseBackend):
             self.options.update(config)
 
     def _prepare_client_options(self):
-            if pymongo.version_tuple >= (3,):
-                return {'maxPoolSize': self.max_pool_size}
-            else:  # pragma: no cover
-                return {'max_pool_size': self.max_pool_size,
-                        'auto_start_request': False}
+        if pymongo.version_tuple >= (3,):
+            return {'maxPoolSize': self.max_pool_size}
+        else:  # pragma: no cover
+            return {'max_pool_size': self.max_pool_size,
+                    'auto_start_request': False}
 
     def _get_connection(self):
         """Connect to the MongoDB server."""

celery/bin/base.py

@@ -79,7 +79,7 @@ def _add_optparse_argument(parser, opt, typemap={
     # store_true sets value to "('NO', 'DEFAULT')" for some
     # crazy reason, so not to set a sane default here.
     if opt.action == 'store_true' and opt.default is None:
-            opt.default = False
+        opt.default = False
     parser.add_argument(
         *opt._long_opts + opt._short_opts,
         **dictfilter(dict(

celery/canvas.py

@@ -1364,6 +1364,8 @@ def signature(varies, *args, **kwargs):
             return varies.clone()
         return Signature.from_dict(varies, app=app)
     return Signature(varies, *args, **kwargs)
+
+
 subtask = signature  # noqa: E305 XXX compat
 
 
@@ -1392,4 +1394,6 @@ def maybe_signature(d, app=None, clone=False):
         if app is not None:
             d._app = app
     return d
+
+
 maybe_subtask = maybe_signature  # noqa: E305 XXX compat

celery/events/state.py

@@ -100,6 +100,8 @@ class CallableDefaultdict(defaultdict):
 
     def __call__(self, *args, **kwargs):
         return self.fun(*args, **kwargs)
+
+
 Callable.register(CallableDefaultdict)  # noqa: E305
 
 

celery/exceptions.py

@@ -161,6 +161,8 @@ class Retry(TaskPredicate):
 
     def __reduce__(self):
         return self.__class__, (self.message, self.excs, self.when)
+
+
 RetryTaskError = Retry  # noqa: E305 XXX compat
 
 
@@ -244,6 +246,8 @@ class CDeprecationWarning(DeprecationWarning):
 
 class WorkerTerminate(SystemExit):
     """Signals that the worker should terminate immediately."""
+
+
 SystemTerminate = WorkerTerminate  # noqa: E305 XXX compat
 
 

celery/platforms.py

@@ -228,6 +228,8 @@ class Pidfile(object):
                     "Inconsistency: Pidfile content doesn't match at re-read")
         finally:
             rfh.close()
+
+
 PIDFile = Pidfile  # noqa: E305 XXX compat alias
 
 

celery/schedules.py

@@ -474,15 +474,20 @@ class crontab(BaseSchedule):
                 return True
             return False
 
+        def is_before_last_run(year, month, day):
+            return self.maybe_make_aware(datetime(year,
+                                                  month,
+                                                  day)) < last_run_at
+
         def roll_over():
             for _ in range(2000):
                 flag = (datedata.dom == len(days_of_month) or
                         day_out_of_range(datedata.year,
                                          months_of_year[datedata.moy],
                                          days_of_month[datedata.dom]) or
-                        (self.maybe_make_aware(datetime(datedata.year,
-                         months_of_year[datedata.moy],
-                         days_of_month[datedata.dom])) < last_run_at))
+                        (is_before_last_run(datedata.year,
+                                            months_of_year[datedata.moy],
+                                            days_of_month[datedata.dom])))
 
                 if flag:
                     datedata.dom = 0

celery/utils/collections.py

@@ -228,6 +228,8 @@ class DictAttribute(object):
         def values(self):
             # type: () -> List[Any]
             return list(self._iterate_values())
+
+
 MutableMapping.register(DictAttribute)  # noqa: E305
 
 
@@ -706,6 +708,8 @@ class LimitedSet(object):
         # type: () -> float
         """Compute how much is heap bigger than data [percents]."""
         return len(self._heap) * 100 / max(len(self._data), 1) - 100
+
+
 MutableSet.register(LimitedSet)  # noqa: E305
 
 
@@ -808,6 +812,8 @@ class Messagebuffer(Evictable):
     def _evictcount(self):
         # type: () -> int
         return len(self)
+
+
 Sequence.register(Messagebuffer)  # noqa: E305
 
 

celery/utils/time.py

@@ -359,7 +359,7 @@ class ffwd(object):
         month = self.month or other.month
         day = min(monthrange(year, month)[1], self.day or other.day)
         ret = other.replace(**dict(dictfilter(self._fields()),
-                            year=year, month=month, day=day))
+                                   year=year, month=month, day=day))
         if self.weekday is not None:
             ret += timedelta(days=(7 - ret.weekday() + self.weekday) % 7)
         return ret + timedelta(days=self.days)

celery/worker/request.py

@@ -50,6 +50,8 @@ def __optimize__():
     global _does_info
     _does_debug = logger.isEnabledFor(logging.DEBUG)
     _does_info = logger.isEnabledFor(logging.INFO)
+
+
 __optimize__()  # noqa: E305
 
 # Localize

celery/worker/worker.py

@@ -43,8 +43,6 @@ except ImportError:  # pragma: no cover
     resource = None  # noqa
 
 
-
-
 __all__ = ['WorkController']
 
 #: Default socket timeout at shutdown.

setup.cfg

@@ -10,7 +10,7 @@ all_files = 1
 [flake8]
 # classes can be lowercase, arguments and variables can be uppercase
 # whenever it makes the code more readable.
-ignore = N806, N802, N801, N803
+ignore = N806, N802, N801, N803, E741, E742, E722
 
 [pep257]
 ignore = D102,D104,D203,D105,D213

t/unit/app/test_beat.py

@@ -312,7 +312,7 @@ class test_Scheduler:
         scheduler = mScheduler(app=self.app)
         nums = [600, 300, 650, 120, 250, 36]
         s = dict(('test_ticks%s' % i,
-                 {'schedule': mocked_schedule(False, j)})
+                  {'schedule': mocked_schedule(False, j)})
                  for i, j in enumerate(nums))
         scheduler.update_from_dict(s)
         assert scheduler.tick() == min(nums) - 0.010

t/unit/apps/test_multi.py

@@ -104,7 +104,7 @@ class test_multi_args:
         assert expand('%h') == '*P*jerry@*S*'
         assert expand('%n') == '*P*jerry'
         nodes2 = list(multi_args(p, cmd='COMMAND', append='',
-                      prefix='*P*', suffix='*S*'))
+                                 prefix='*P*', suffix='*S*'))
         assert nodes2[0].argv[-1] == '-- .disable_rate_limits=1'
 
         p2 = NamespacedOptionParser(['10', '-c:1', '5'])

t/unit/backends/test_couchdb.py

@@ -33,10 +33,10 @@ class test_CouchBackend:
             module.pycouchdb = prev
 
     def test_get_container_exists(self):
-            self.backend._connection = sentinel._connection
-            connection = self.backend.connection
-            assert connection is sentinel._connection
-            self.Server.assert_not_called()
+        self.backend._connection = sentinel._connection
+        connection = self.backend.connection
+        assert connection is sentinel._connection
+        self.Server.assert_not_called()
 
     def test_get(self):
         """test_get

t/unit/backends/test_dynamodb.py

@@ -43,37 +43,37 @@ class test_DynamoDBBackend:
             'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'
         with patch('boto3.client') as mock_boto_client, \
                 patch(table_creation_path):
-                backend = DynamoDBBackend(
-                    app=self.app,
-                    url='dynamodb://@localhost:8000'
-                )
-                client = backend._get_client()
-                assert backend.client is client
-                mock_boto_client.assert_called_once_with(
-                    'dynamodb',
-                    endpoint_url='http://localhost:8000',
-                    region_name='us-east-1'
-                )
-                assert backend.endpoint_url == 'http://localhost:8000'
+            backend = DynamoDBBackend(
+                app=self.app,
+                url='dynamodb://@localhost:8000'
+            )
+            client = backend._get_client()
+            assert backend.client is client
+            mock_boto_client.assert_called_once_with(
+                'dynamodb',
+                endpoint_url='http://localhost:8000',
+                region_name='us-east-1'
+            )
+            assert backend.endpoint_url == 'http://localhost:8000'
 
     def test_get_client_credentials(self):
         table_creation_path = \
             'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'
         with patch('boto3.client') as mock_boto_client, \
                 patch(table_creation_path):
-                backend = DynamoDBBackend(
-                    app=self.app,
-                    url='dynamodb://key:secret@test'
-                )
-                client = backend._get_client()
-                assert client is backend.client
-                mock_boto_client.assert_called_once_with(
-                    'dynamodb',
-                    aws_access_key_id='key',
-                    aws_secret_access_key='secret',
-                    region_name='test'
-                )
-                assert backend.aws_region == 'test'
+            backend = DynamoDBBackend(
+                app=self.app,
+                url='dynamodb://key:secret@test'
+            )
+            client = backend._get_client()
+            assert client is backend.client
+            mock_boto_client.assert_called_once_with(
+                'dynamodb',
+                aws_access_key_id='key',
+                aws_secret_access_key='secret',
+                region_name='test'
+            )
+            assert backend.aws_region == 'test'
 
     def test_get_or_create_table_not_exists(self):
         self.backend._client = MagicMock()

t/unit/bin/test_events.py

@@ -34,6 +34,8 @@ class MockCommand(object):
 
 def proctitle(prog, info=None):
     proctitle.last = (prog, info)
+
+
 proctitle.last = ()  # noqa: E305
 
 

t/unit/bin/test_worker.py

@@ -406,7 +406,7 @@ class test_funs:
         cmd = worker()
         cmd.app = self.app
         opts, args = cmd.parse_options('worker', ['--concurrency=512',
-                                       '--heartbeat-interval=10'])
+                                                  '--heartbeat-interval=10'])
         assert opts['concurrency'] == 512
         assert opts['heartbeat_interval'] == 10
 

t/unit/events/test_state.py

@@ -153,7 +153,7 @@ class ev_snapshot(replay):
             worker = not i % 2 and 'utest2' or 'utest1'
             type = not i % 2 and 'task2' or 'task1'
             self.events.append(Event('task-received', name=type,
-                               uuid=uuid(), hostname=worker))
+                                     uuid=uuid(), hostname=worker))
 
 
 class test_Worker:

t/unit/fixups/test_django.py

@@ -135,22 +135,22 @@ class test_DjangoWorkerFixup(FixupCase):
     def test_on_worker_process_init(self, patching):
         with self.fixup_context(self.app) as (f, _, _):
             with patch('celery.fixups.django._maybe_close_fd') as mcf:
-                    _all = f._db.connections.all = Mock()
-                    conns = _all.return_value = [
-                        Mock(), Mock(),
-                    ]
-                    conns[0].connection = None
-                    with patch.object(f, 'close_cache'):
-                        with patch.object(f, '_close_database'):
-                            f.on_worker_process_init()
-                            mcf.assert_called_with(conns[1].connection)
-                            f.close_cache.assert_called_with()
-                            f._close_database.assert_called_with()
-
-                            f.validate_models = Mock(name='validate_models')
-                            patching.setenv('FORKED_BY_MULTIPROCESSING', '1')
-                            f.on_worker_process_init()
-                            f.validate_models.assert_called_with()
+                _all = f._db.connections.all = Mock()
+                conns = _all.return_value = [
+                    Mock(), Mock(),
+                ]
+                conns[0].connection = None
+                with patch.object(f, 'close_cache'):
+                    with patch.object(f, '_close_database'):
+                        f.on_worker_process_init()
+                        mcf.assert_called_with(conns[1].connection)
+                        f.close_cache.assert_called_with()
+                        f._close_database.assert_called_with()
+
+                        f.validate_models = Mock(name='validate_models')
+                        patching.setenv('FORKED_BY_MULTIPROCESSING', '1')
+                        f.on_worker_process_init()
+                        f.validate_models.assert_called_with()
 
     def test_on_task_prerun(self):
         task = Mock()

t/unit/tasks/test_result.py

@@ -534,17 +534,17 @@ class MockAsyncResultSuccess(AsyncResult):
 
 
 class SimpleBackend(SyncBackendMixin):
-        ids = []
+    ids = []
 
-        def __init__(self, ids=[]):
-            self.ids = ids
+    def __init__(self, ids=[]):
+        self.ids = ids
 
-        def _ensure_not_eager(self):
-            pass
+    def _ensure_not_eager(self):
+        pass
 
-        def get_many(self, *args, **kwargs):
-            return ((id, {'result': i, 'status': states.SUCCESS})
-                    for i, id in enumerate(self.ids))
+    def get_many(self, *args, **kwargs):
+        return ((id, {'result': i, 'status': states.SUCCESS})
+                for i, id in enumerate(self.ids))
 
 
 class test_GroupResult:

t/unit/utils/test_utils.py

@@ -11,8 +11,8 @@ from celery.utils import cached_property, chunks
     (range(10), 2, [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]),
 ])
 def test_chunks(items, n, expected):
-        x = chunks(iter(list(items)), n)
-        assert list(x) == expected
+    x = chunks(iter(list(items)), n)
+    assert list(x) == expected
 
 
 def test_cached_property():