pyupgrade.

celery/app/autoretry.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """Tasks auto-retry functionality."""
 from vine.utils import wraps
 

celery/backends/database/session.py

@@ -36,8 +36,8 @@ def get_engine(self, dburi, **kwargs):
                 engine = self._engines[dburi] = create_engine(dburi, **kwargs)
                 return engine
         else:
-            kwargs = dict([(k, v) for k, v in kwargs.items() if
-                           not k.startswith('pool')])
+            kwargs = {k: v for k, v in kwargs.items() if
+                      not k.startswith('pool')}
             return create_engine(dburi, poolclass=NullPool, **kwargs)
 
     def create_session(self, dburi, short_lived_sessions=False, **kwargs):

celery/backends/filesystem.py

@@ -8,13 +8,6 @@
 from celery.backends.base import KeyValueStoreBackend
 from celery.exceptions import ImproperlyConfigured
 
-# Python 2 does not have FileNotFoundError and IsADirectoryError
-try:
-    FileNotFoundError
-except NameError:
-    FileNotFoundError = IOError
-    IsADirectoryError = IOError
-
 default_encoding = locale.getpreferredencoding(False)
 
 E_NO_PATH_SET = 'You need to configure a path for the file-system backend'
@@ -58,7 +51,7 @@ def __init__(self, url=None, open=open, unlink=os.unlink, sep=os.sep,
     def __reduce__(self, args=(), kwargs={}):
         kwargs.update(
             dict(url=self.url))
-        return super(FilesystemBackend, self).__reduce__(args, kwargs)
+        return super().__reduce__(args, kwargs)
 
     def _find_path(self, url):
         if not url:

celery/bin/amqp.py

@@ -171,7 +171,7 @@ def queue_declare(amqp_context, queue, passive, durable, auto_delete):
             amqp_context.reconnect()
         else:
             amqp_context.cli_context.secho(
-                'queue:{0} messages:{1} consumers:{2}'.format(*retval),
+                'queue:{} messages:{} consumers:{}'.format(*retval),
                 fg='cyan', bold=True)
             amqp_context.echo_ok()
 

celery/bin/base.py

@@ -120,13 +120,13 @@ class CeleryOption(click.Option):
     def get_default(self, ctx):
         if self.default_value_from_context:
             self.default = ctx.obj[self.default_value_from_context]
-        return super(CeleryOption, self).get_default(ctx)
+        return super().get_default(ctx)
 
     def __init__(self, *args, **kwargs):
         """Initialize a Celery option."""
         self.help_group = kwargs.pop('help_group', None)
         self.default_value_from_context = kwargs.pop('default_value_from_context', None)
-        super(CeleryOption, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
 
 
 class CeleryCommand(click.Command):

celery/bin/control.py

@@ -30,7 +30,7 @@ def _consume_arguments(meta, method, args):
                 if meta.variadic:
                     break
                 raise click.UsageError(
-                    'Command {0!r} takes arguments: {1}'.format(
+                    'Command {!r} takes arguments: {}'.format(
                         method, meta.signature))
             else:
                 yield name, typ(arg) if typ is not None else arg
@@ -86,7 +86,7 @@ def status(ctx, timeout, destination, json, **kwargs):
         ctx.obj.echo(dumps(replies))
     nodecount = len(replies)
     if not kwargs.get('quiet', False):
-        ctx.obj.echo('\n{0} {1} online.'.format(
+        ctx.obj.echo('\n{} {} online.'.format(
             nodecount, text.pluralize(nodecount, 'node')))
 
 
@@ -134,7 +134,7 @@ def inspect(ctx, action, timeout, destination, json, **kwargs):
         ctx.obj.echo(dumps(replies))
     nodecount = len(replies)
     if not ctx.obj.quiet:
-        ctx.obj.echo('\n{0} {1} online.'.format(
+        ctx.obj.echo('\n{} {} online.'.format(
             nodecount, text.pluralize(nodecount, 'node')))
 
 

celery/bin/events.py

@@ -9,8 +9,8 @@
 
 
 def _set_process_status(prog, info=''):
-    prog = '{0}:{1}'.format('celery events', prog)
-    info = '{0} {1}'.format(info, strargv(sys.argv))
+    prog = '{}:{}'.format('celery events', prog)
+    info = '{} {}'.format(info, strargv(sys.argv))
     return set_process_title(prog, info=info)
 
 

celery/bin/graph.py

@@ -42,10 +42,10 @@ def maybe_list(l, sep=','):
     generic = 'generic' in args
 
     def generic_label(node):
-        return '{0} ({1}://)'.format(type(node).__name__,
+        return '{} ({}://)'.format(type(node).__name__,
                                      node._label.split('://')[0])
 
-    class Node(object):
+    class Node:
         force_label = None
         scheme = {}
 
@@ -71,8 +71,8 @@ class Thread(Node):
 
         def __init__(self, label, **kwargs):
             self.real_label = label
-            super(Thread, self).__init__(
-                label='thr-{0}'.format(next(tids)),
+            super().__init__(
+                label='thr-{}'.format(next(tids)),
                 pos=0,
             )
 
@@ -139,11 +139,11 @@ def maybe_abbr(l, name, max=Wmax):
         size = len(l)
         abbr = max and size > max
         if 'enumerate' in args:
-            l = ['{0}{1}'.format(name, subscript(i + 1))
+            l = ['{}{}'.format(name, subscript(i + 1))
                  for i, obj in enumerate(l)]
         if abbr:
             l = l[0:max - 1] + [l[size - 1]]
-            l[max - 2] = '{0}⎨…{1}⎬'.format(
+            l[max - 2] = '{}⎨…{}⎬'.format(
                 name[0], subscript(size - (max - 1)))
         return l
 

celery/bin/list.py

@@ -29,7 +29,7 @@ def bindings(ctx):
             raise click.UsageError('Your transport cannot list bindings.')
 
         def fmt(q, e, r):
-            ctx.obj.echo('{0:<28} {1:<28} {2}'.format(q, e, r))
+            ctx.obj.echo(f'{q:<28} {e:<28} {r}')
         fmt('Queue', 'Exchange', 'Routing Key')
         fmt('-' * 16, '-' * 16, '-' * 16)
         for b in bindings:

celery/bin/logtool.py

@@ -32,15 +32,15 @@ class _task_counts(list):
 
     @property
     def format(self):
-        return '\n'.join('{0}: {1}'.format(*i) for i in self)
+        return '\n'.join('{}: {}'.format(*i) for i in self)
 
 
 def task_info(line):
     m = RE_TASK_INFO.match(line)
     return m.groups()
 
 
-class Audit(object):
+class Audit:
 
     def __init__(self, on_task_error=None, on_trace=None, on_debug=None):
         self.ids = set()

celery/bin/multi.py

@@ -167,7 +167,7 @@ def _inner(self, *argv, **kwargs):
     return _inner
 
 
-class TermLogger(object):
+class TermLogger:
 
     splash_text = 'celery multi v{version}'
     splash_context = {'version': VERSION_BANNER}
@@ -277,7 +277,7 @@ def call_command(self, command, argv):
         try:
             return self.commands[command](*argv) or EX_OK
         except KeyError:
-            return self.error('Invalid command: {0}'.format(command))
+            return self.error(f'Invalid command: {command}')
 
     def _handle_reserved_options(self, argv):
         argv = list(argv)  # don't modify callers argv.
@@ -402,7 +402,7 @@ def on_still_waiting_for(self, nodes):
         num_left = len(nodes)
         if num_left:
             self.note(self.colored.blue(
-                '> Waiting for {0} {1} -> {2}...'.format(
+                '> Waiting for {} {} -> {}...'.format(
                     num_left, pluralize(num_left, 'node'),
                     ', '.join(str(node.pid) for node in nodes)),
             ), newline=False)
@@ -419,17 +419,17 @@ def on_node_signal_dead(self, node):
                 node))
 
     def on_node_start(self, node):
-        self.note('\t> {0.name}: '.format(node), newline=False)
+        self.note(f'\t> {node.name}: ', newline=False)
 
     def on_node_restart(self, node):
         self.note(self.colored.blue(
-            '> Restarting node {0.name}: '.format(node)), newline=False)
+            f'> Restarting node {node.name}: '), newline=False)
 
     def on_node_down(self, node):
-        self.note('> {0.name}: {1.DOWN}'.format(node, self))
+        self.note(f'> {node.name}: {self.DOWN}')
 
     def on_node_shutdown_ok(self, node):
-        self.note('\n\t> {0.name}: {1.OK}'.format(node, self))
+        self.note(f'\n\t> {node.name}: {self.OK}')
 
     def on_node_status(self, node, retval):
         self.note(retval and self.FAILED or self.OK)
@@ -439,13 +439,13 @@ def on_node_signal(self, node, sig):
             node, sig=sig))
 
     def on_child_spawn(self, node, argstr, env):
-        self.info('  {0}'.format(argstr))
+        self.info(f'  {argstr}')
 
     def on_child_signalled(self, node, signum):
-        self.note('* Child was terminated by signal {0}'.format(signum))
+        self.note(f'* Child was terminated by signal {signum}')
 
     def on_child_failure(self, node, retcode):
-        self.note('* Child terminated with exit code {0}'.format(retcode))
+        self.note(f'* Child terminated with exit code {retcode}')
 
     @cached_property
     def OK(self):

celery/bin/upgrade.py

@@ -30,7 +30,7 @@ def _compat_key(key, namespace='CELERY'):
 def _backup(filename, suffix='.orig'):
     lines = []
     backup_filename = ''.join([filename, suffix])
-    print('writing backup to {0}...'.format(backup_filename),
+    print(f'writing backup to {backup_filename}...',
           file=sys.stderr)
     with codecs.open(filename, 'r', 'utf-8') as read_fh:
         with codecs.open(backup_filename, 'w', 'utf-8') as backup_fh:
@@ -71,7 +71,7 @@ def settings(filename, django, compat, no_backup):
     """Migrate settings from Celery 3.x to Celery 4.x."""
     lines = _slurp(filename)
     keyfilter = _compat_key if django or compat else pass1
-    print('processing {0}...'.format(filename), file=sys.stderr)
+    print(f'processing {filename}...', file=sys.stderr)
     # gives list of tuples: ``(did_change, line_contents)``
     new_lines = [
         _to_new_key(line, keyfilter) for line in lines

celery/contrib/testing/mocks.py

@@ -5,10 +5,7 @@
 try:
     from case import Mock
 except ImportError:
-    try:
-        from unittest.mock import Mock
-    except ImportError:
-        from mock import Mock
+    from unittest.mock import Mock
 
 
 def TaskMessage(

celery/security/__init__.py

@@ -64,8 +64,8 @@ def setup_security(allowed_serializers=None, key=None, cert=None, store=None,
     if not (key and cert and store):
         raise ImproperlyConfigured(SECURITY_SETTING_MISSING)
 
-    with open(key, 'r') as kf:
-        with open(cert, 'r') as cf:
+    with open(key) as kf:
+        with open(cert) as cf:
             register_auth(kf.read(), cf.read(), store, digest, serializer)
     registry._set_default_serializer('auth')
 

t/benchmarks/bench_worker.py

@@ -55,7 +55,7 @@ def it(_, n):
     elif i > n - 2:
         total = tdiff(it.time_start)
         print('({} so far: {}s)'.format(i, tdiff(it.subt)), file=sys.stderr)
-        print('-- process {0} tasks: {1}s total, {2} tasks/s'.format(
+        print('-- process {} tasks: {}s total, {} tasks/s'.format(
             n, total, n / (total + .0),
         ))
         import os

t/unit/apps/test_multi.py

@@ -116,7 +116,7 @@ def _args(name, *args):
             return args + (
                 '--pidfile={}.pid'.format(os.path.join(os.path.normpath('/var/run/celery/'), name)),
                 '--logfile={}%I.log'.format(os.path.join(os.path.normpath('/var/log/celery/'), name)),
-                '--executable={0}'.format(sys.executable),
+                f'--executable={sys.executable}',
                 '',
             )
 
@@ -406,7 +406,7 @@ def test_getpids(self):
         assert node_0.name == 'foo@e.com'
         assert sorted(node_0.argv) == sorted([
             '',
-            '--executable={0}'.format(node_0.executable),
+            f'--executable={node_0.executable}',
             '--logfile={}'.format(os.path.normpath('/var/log/celery/foo%I.log')),
             '--pidfile={}'.format(os.path.normpath('/var/run/celery/foo.pid')),
             '-m celery worker --detach',
@@ -417,7 +417,7 @@ def test_getpids(self):
         assert node_1.name == 'bar@e.com'
         assert sorted(node_1.argv) == sorted([
             '',
-            '--executable={0}'.format(node_1.executable),
+            f'--executable={node_1.executable}',
             '--logfile={}'.format(os.path.normpath('/var/log/celery/bar%I.log')),
             '--pidfile={}'.format(os.path.normpath('/var/run/celery/bar.pid')),
             '-m celery worker --detach',

t/unit/backends/test_asynchronous.py

@@ -20,7 +20,7 @@ def setup_eventlet():
     os.environ.update(EVENTLET_NO_GREENDNS='yes')
 
 
-class DrainerTests(object):
+class DrainerTests:
     """
     Base test class for the Default / Gevent / Eventlet drainers.
     """

t/unit/backends/test_cassandra.py

@@ -149,7 +149,7 @@ def __init__(self, *args, **kwargs):
             def execute(self, *args, **kwargs):
                 raise OTOExc()
 
-        class DummyCluster(object):
+        class DummyCluster:
 
             def __init__(self, *args, **kwargs):
                 pass
@@ -170,7 +170,7 @@ def test_init_session(self):
         # Tests behavior when Cluster.connect works properly
         from celery.backends import cassandra as mod
 
-        class DummyCluster(object):
+        class DummyCluster:
 
             def __init__(self, *args, **kwargs):
                 pass

t/unit/backends/test_mongodb.py

@@ -568,7 +568,7 @@ def test_encode_decode(self, mongo_backend_factory, serializer,
         assert decoded == 12
 
 
-class _MyTestClass(object):
+class _MyTestClass:
 
     def __init__(self, a):
         self.a = a
@@ -632,7 +632,7 @@ def fake_mongo_collection_patch(self, monkeypatch):
         """A fake collection with serialization experience close to MongoDB."""
         bson = pytest.importorskip("bson")
 
-        class FakeMongoCollection(object):
+        class FakeMongoCollection:
             def __init__(self):
                 self.data = {}