[pull] master from celery:master

.github/workflows/python-package.yml

@@ -29,10 +29,10 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', 'pypy-3.7', 'pypy-3.8']
+        python-version: ['3.7', '3.8', '3.9', '3.10', 'pypy-3.9', 'pypy-3.8']
         os: ["ubuntu-latest", "windows-latest"]
         exclude:
-            - python-version: 'pypy-3.7'
+            - python-version: 'pypy-3.9'
               os: "windows-latest"
             - python-version: 'pypy-3.8'
               os: "windows-latest"
@@ -120,7 +120,7 @@ jobs:
               run: |
                   echo "::set-output name=dir::$(pip cache dir)"
           -   name: Install tox
-              run: python -m pip install tox
+              run: python -m pip install --upgrade pip tox tox-gh-actions
           -   name: >
                   Run tox for
                   "${{ matrix.python-version }}-integration-${{ matrix.toxenv }}"

.pre-commit-config.yaml

@@ -1,12 +1,12 @@
 repos:
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.2.0
+    rev: v3.3.1
     hooks:
       - id: pyupgrade
         args: ["--py37-plus"]
 
   - repo: https://github.com/PyCQA/flake8
-    rev: 5.0.4
+    rev: 6.0.0
     hooks:
       - id: flake8
 
@@ -16,20 +16,20 @@ repos:
       - id: yesqa
 
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.3.0
+    rev: v4.4.0
     hooks:
       - id: check-merge-conflict
       - id: check-toml
       - id: check-yaml
       - id: mixed-line-ending
 
   - repo: https://github.com/pycqa/isort
-    rev: 5.10.1
+    rev: v5.11.3
     hooks:
       - id: isort
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.982
+    rev: v0.991
     hooks:
     -   id: mypy
         pass_filenames: false

CONTRIBUTORS.txt

@@ -290,3 +290,5 @@ Gabor Boros, 2021/11/09
 Tizian Seehaus, 2022/02/09
 Oleh Romanovskyi, 2022/06/09
 JoonHwan Kim, 2022/08/01
+Kaustav Banerjee, 2022/11/10
+Austin Snoeyink 2022/12/06

celery/__init__.py

@@ -70,8 +70,7 @@ def debug_import(name, locals=None, globals=None,
     from celery.app.base import Celery
     from celery.app.task import Task
     from celery.app.utils import bugreport
-    from celery.canvas import (chain, chord, chunks, group, maybe_signature, signature, subtask, xmap,  # noqa
-                               xstarmap)
+    from celery.canvas import chain, chord, chunks, group, maybe_signature, signature, subtask, xmap, xstarmap
     from celery.utils import uuid
 
 # Eventlet/gevent patching must happen before importing

celery/app/amqp.py

@@ -321,7 +321,7 @@ def as_task_v2(self, task_id, name, args=None, kwargs=None,
         if not root_id:  # empty root_id defaults to task_id
             root_id = task_id
 
-        stamps = {header: maybe_list(options[header]) for header in stamped_headers or []}
+        stamps = {header: options[header] for header in stamped_headers or []}
         headers = {
             'lang': 'py',
             'task': name,

celery/app/base.py

@@ -33,8 +33,7 @@
 from celery.utils.time import maybe_make_aware, timezone, to_utc
 
 # Load all builtin tasks
-from . import builtins  # noqa
-from . import backends
+from . import backends, builtins
 from .annotations import prepare as prepare_annotations
 from .autoretry import add_autoretry_behaviour
 from .defaults import DEFAULT_SECURITY_DIGEST, find_deprecated_settings
@@ -458,6 +457,9 @@ def cons(app):
                     sum([len(args), len(opts)])))
         return inner_create_task_cls(**opts)
 
+    def type_checker(self, fun, bound=False):
+        return staticmethod(head_from_fun(fun, bound=bound))
+
     def _task_from_fun(self, fun, name=None, base=None, bind=False, **options):
         if not self.finalized and not self.autofinalize:
             raise RuntimeError('Contract breach: app not finalized')
@@ -474,7 +476,7 @@ def _task_from_fun(self, fun, name=None, base=None, bind=False, **options):
                 '__doc__': fun.__doc__,
                 '__module__': fun.__module__,
                 '__annotations__': fun.__annotations__,
-                '__header__': staticmethod(head_from_fun(fun, bound=bind)),
+                '__header__': self.type_checker(fun, bound=bind),
                 '__wrapped__': run}, **options))()
             # for some reason __qualname__ cannot be set in type()
             # so we have to set it here.
@@ -778,6 +780,10 @@ def send_task(self, name, args=None, kwargs=None, countdown=None,
             **options
         )
 
+        stamped_headers = options.pop('stamped_headers', [])
+        for stamp in stamped_headers:
+            options.pop(stamp)
+
         if connection:
             producer = amqp.Producer(connection, auto_declare=False)
 

celery/app/defaults.py

@@ -78,6 +78,7 @@ def __repr__(self):
         scheduler=Option('celery.beat:PersistentScheduler'),
         schedule_filename=Option('celerybeat-schedule'),
         sync_every=Option(0, type='int'),
+        cron_starting_deadline=Option(None, type=int)
     ),
     broker=Namespace(
         url=Option(None, type='string'),
@@ -89,6 +90,7 @@ def __repr__(self):
         connection_retry=Option(True, type='bool'),
         connection_retry_on_startup=Option(None, type='bool'),
         connection_max_retries=Option(100, type='int'),
+        channel_error_retry=Option(False, type='bool'),
         failover_strategy=Option(None, type='string'),
         heartbeat=Option(120, type='int'),
         heartbeat_checkrate=Option(3.0, type='int'),

celery/app/task.py

@@ -953,10 +953,18 @@ def replace(self, sig):
         for t in reversed(self.request.chain or []):
             sig |= signature(t, app=self.app)
         # Stamping sig with parents groups
-        stamped_headers = self.request.stamped_headers
         if self.request.stamps:
             groups = self.request.stamps.get("groups")
-            sig.stamp(visitor=GroupStampingVisitor(groups=groups, stamped_headers=stamped_headers))
+            sig.stamp(visitor=GroupStampingVisitor(groups=groups, stamped_headers=self.request.stamped_headers))
+            stamped_headers = self.request.stamped_headers.copy()
+            stamps = self.request.stamps.copy()
+            stamped_headers.extend(sig.options.get('stamped_headers', []))
+            stamps.update({
+                stamp: value
+                for stamp, value in sig.options.items() if stamp in sig.options.get('stamped_headers', [])
+            })
+            sig.options['stamped_headers'] = stamped_headers
+            sig.options.update(stamps)
 
         return self.on_replace(sig)
 

celery/app/trace.py

@@ -10,7 +10,7 @@
 from collections import namedtuple
 from warnings import warn
 
-from billiard.einfo import ExceptionInfo
+from billiard.einfo import ExceptionInfo, ExceptionWithTraceback
 from kombu.exceptions import EncodeError
 from kombu.serialization import loads as loads_message
 from kombu.serialization import prepare_accept_content
@@ -238,6 +238,8 @@ def handle_failure(self, task, req, store_errors=True, call_errbacks=True):
 
     def _log_error(self, task, req, einfo):
         eobj = einfo.exception = get_pickled_exception(einfo.exception)
+        if isinstance(eobj, ExceptionWithTraceback):
+            eobj = einfo.exception = eobj.exc
         exception, traceback, exc_info, sargs, skwargs = (
             safe_repr(eobj),
             safe_str(einfo.traceback),

celery/backends/base.py

@@ -397,7 +397,7 @@ def exception_to_python(self, exc):
                 exc = cls(*exc_msg)
             else:
                 exc = cls(exc_msg)
-        except Exception as err:  # noqa
+        except Exception as err:
             exc = Exception(f'{cls}({exc_msg})')
 
         return exc
@@ -817,11 +817,25 @@ class BaseKeyValueStoreBackend(Backend):
     def __init__(self, *args, **kwargs):
         if hasattr(self.key_t, '__func__'):  # pragma: no cover
             self.key_t = self.key_t.__func__  # remove binding
-        self._encode_prefixes()
         super().__init__(*args, **kwargs)
+        self._add_global_keyprefix()
+        self._encode_prefixes()
         if self.implements_incr:
             self.apply_chord = self._apply_chord_incr
 
+    def _add_global_keyprefix(self):
+        """
+        This method prepends the global keyprefix to the existing keyprefixes.
+
+        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
+        `global_keyprefix` key. If so, then it is prepended to the task, group and chord key prefixes.
+        """
+        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)
+        if global_keyprefix:
+            self.task_keyprefix = f"{global_keyprefix}_{self.task_keyprefix}"
+            self.group_keyprefix = f"{global_keyprefix}_{self.group_keyprefix}"
+            self.chord_keyprefix = f"{global_keyprefix}_{self.chord_keyprefix}"
+
     def _encode_prefixes(self):
         self.task_keyprefix = self.key_t(self.task_keyprefix)
         self.group_keyprefix = self.key_t(self.group_keyprefix)

celery/beat.py

@@ -46,7 +46,7 @@ class SchedulingError(Exception):
 
 
 class BeatLazyFunc:
-    """An lazy function declared in 'beat_schedule' and called before sending to worker.
+    """A lazy function declared in 'beat_schedule' and called before sending to worker.
 
     Example:
 

celery/bin/shell.py

@@ -67,10 +67,10 @@ def _no_ipython(self):  # pragma: no cover
 
 def _invoke_default_shell(locals):
     try:
-        import IPython  # noqa
+        import IPython
     except ImportError:
         try:
-            import bpython  # noqa
+            import bpython
         except ImportError:
             _invoke_fallback_shell(locals)
         else: