Repository: jrobichaud/django-structlog

Issue:
Add queue information to task_enqueued log
Is there a way to add queue information to this log?
```
    logger.info( 
         "task_enqueued", 
         child_task_id=headers.get("id") if headers else body.get("id"), 
         child_task_name=headers.get("task") if headers else body.get("task"), 
     )
```

Something like
````
    logger.info( 
         "task_enqueued", 
         child_task_id=headers.get("id") if headers else body.get("id"), 
         child_task_name=headers.get("task") if headers else body.get("task"), 
         priority=,
         queue=
     )
````

Discussion:

Patch:
diff --git a/django_structlog/apps.py b/django_structlog/apps.py
index 50f2958..726205a 100644
--- a/django_structlog/apps.py
+++ b/django_structlog/apps.py
@@ -8,11 +8,13 @@ class DjangoStructLogConfig(AppConfig):
 
     def ready(self):
         if app_settings.CELERY_ENABLED:
-            from .celery.receivers import connect_celery_signals
+            from .celery.receivers import CeleryReceiver
 
-            connect_celery_signals()
+            self._celery_receiver = CeleryReceiver()
+            self._celery_receiver.connect_signals()
 
         if app_settings.COMMAND_LOGGING_ENABLED:
-            from .commands import init_command_signals
+            from .commands import DjangoCommandReceiver
 
-            init_command_signals()
+            self._django_command_receiver = DjangoCommandReceiver()
+            self._django_command_receiver.connect_signals()
diff --git a/django_structlog/celery/receivers.py b/django_structlog/celery/receivers.py
index 017a794..43ad50c 100644
--- a/django_structlog/celery/receivers.py
+++ b/django_structlog/celery/receivers.py
@@ -6,107 +6,155 @@ from . import signals
 logger = structlog.getLogger(__name__)
 
 
-def receiver_before_task_publish(sender=None, headers=None, body=None, **kwargs):
-    import celery
-
-    if celery.current_app.conf.task_protocol < 2:
-        return
-
-    context = structlog.contextvars.get_merged_contextvars(logger)
-    if "task_id" in context:
-        context["parent_task_id"] = context.pop("task_id")
-
-    signals.modify_context_before_task_publish.send(
-        sender=receiver_before_task_publish, context=context
-    )
-
-    headers["__django_structlog__"] = context
-
-
-def receiver_after_task_publish(sender=None, headers=None, body=None, **kwargs):
-    logger.info(
-        "task_enqueued",
-        child_task_id=headers.get("id") if headers else body.get("id"),
-        child_task_name=headers.get("task") if headers else body.get("task"),
-    )
-
-
-def receiver_task_pre_run(task_id, task, *args, **kwargs):
-    structlog.contextvars.clear_contextvars()
-    structlog.contextvars.bind_contextvars(task_id=task_id)
-    metadata = getattr(task.request, "__django_structlog__", {})
-    structlog.contextvars.bind_contextvars(**metadata)
-    signals.bind_extra_task_metadata.send(
-        sender=receiver_task_pre_run, task=task, logger=logger
-    )
-    logger.info("task_started", task=task.name)
-
-
-def receiver_task_retry(request=None, reason=None, einfo=None, **kwargs):
-    logger.warning("task_retrying", reason=reason)
-
+class CeleryReceiver:
+    def __init__(self):
+        self._priority = None
+
+    def receiver_before_task_publish(
+        self,
+        sender=None,
+        headers=None,
+        body=None,
+        properties=None,
+        routing_key=None,
+        **kwargs,
+    ):
+        import celery
+
+        if celery.current_app.conf.task_protocol < 2:
+            return
+
+        context = structlog.contextvars.get_merged_contextvars(logger)
+        if "task_id" in context:
+            context["parent_task_id"] = context.pop("task_id")
+
+        signals.modify_context_before_task_publish.send(
+            sender=self.receiver_before_task_publish,
+            context=context,
+            task_routing_key=routing_key,
+            task_properties=properties,
+        )
+        if properties:
+            self._priority = properties.get("priority", None)
 
-def receiver_task_success(result=None, **kwargs):
-    signals.pre_task_succeeded.send(
-        sender=receiver_task_success, logger=logger, result=result
-    )
-    logger.info("task_succeeded")
+        headers["__django_structlog__"] = context
 
+    def receiver_after_task_publish(
+        self, sender=None, headers=None, body=None, routing_key=None, **kwargs
+    ):
+        properties = {}
+        if self._priority is not None:
+            properties["priority"] = self._priority
+            self._priority = None
 
-def receiver_task_failure(
-    task_id=None,
-    exception=None,
-    traceback=None,
-    einfo=None,
-    sender=None,
-    *args,
-    **kwargs,
-):
-    throws = getattr(sender, "throws", ())
-    if isinstance(exception, throws):
         logger.info(
-            "task_failed",
-            error=str(exception),
+            "task_enqueued",
+            child_task_id=headers.get("id") if headers else body.get("id"),
+            child_task_name=headers.get("task") if headers else body.get("task"),
+            routing_key=routing_key,
+            **properties,
         )
-    else:
-        logger.exception(
-            "task_failed",
-            error=str(exception),
-            exception=exception,
-        )
-
 
-def receiver_task_revoked(
-    request=None, terminated=None, signum=None, expired=None, **kwargs
-):
-    metadata = getattr(request, "__django_structlog__", {}).copy()
-    metadata["task_id"] = request.id
-    metadata["task"] = request.task
-
-    logger.warning(
-        "task_revoked",
-        terminated=terminated,
-        signum=signum.value if signum is not None else None,
-        signame=signum.name if signum is not None else None,
-        expired=expired,
-        **metadata,
-    )
+    def receiver_task_prerun(self, task_id, task, *args, **kwargs):
+        structlog.contextvars.clear_contextvars()
+        structlog.contextvars.bind_contextvars(task_id=task_id)
+        metadata = getattr(task.request, "__django_structlog__", {})
+        structlog.contextvars.bind_contextvars(**metadata)
+        signals.bind_extra_task_metadata.send(
+            sender=self.receiver_task_prerun, task=task, logger=logger
+        )
+        logger.info("task_started", task=task.name)
 
+    def receiver_task_retry(self, request=None, reason=None, einfo=None, **kwargs):
+        logger.warning("task_retrying", reason=reason)
 
-def receiver_task_unknown(message=None, exc=None, name=None, id=None, **kwargs):
-    logger.error(
-        "task_not_found",
-        task=name,
-        task_id=id,
-    )
+    def receiver_task_success(self, result=None, **kwargs):
+        signals.pre_task_succeeded.send(
+            sender=self.receiver_task_success, logger=logger, result=result
+        )
+        logger.info("task_succeeded")
+
+    def receiver_task_failure(
+        self,
+        task_id=None,
+        exception=None,
+        traceback=None,
+        einfo=None,
+        sender=None,
+        *args,
+        **kwargs,
+    ):
+        throws = getattr(sender, "throws", ())
+        if isinstance(exception, throws):
+            logger.info(
+                "task_failed",
+                error=str(exception),
+            )
+        else:
+            logger.exception(
+                "task_failed",
+                error=str(exception),
+                exception=exception,
+            )
+
+    def receiver_task_revoked(
+        self, request=None, terminated=None, signum=None, expired=None, **kwargs
+    ):
+        metadata = getattr(request, "__django_structlog__", {}).copy()
+        metadata["task_id"] = request.id
+        metadata["task"] = request.task
+
+        logger.warning(
+            "task_revoked",
+            terminated=terminated,
+            signum=signum.value if signum is not None else None,
+            signame=signum.name if signum is not None else None,
+            expired=expired,
+            **metadata,
+        )
 
+    def receiver_task_unknown(
+        self, message=None, exc=None, name=None, id=None, **kwargs
+    ):
+        logger.error(
+            "task_not_found",
+            task=name,
+            task_id=id,
+        )
 
-def receiver_task_rejected(message=None, exc=None, **kwargs):
-    logger.exception("task_rejected", task_id=message.properties.get("correlation_id"))
+    def receiver_task_rejected(self, message=None, exc=None, **kwargs):
+        logger.exception(
+            "task_rejected", task_id=message.properties.get("correlation_id")
+        )
 
+    def connect_signals(self):
+        from celery.signals import (
+            before_task_publish,
+            after_task_publish,
+        )
 
-def connect_celery_signals():
-    from celery.signals import before_task_publish, after_task_publish
+        before_task_publish.connect(self.receiver_before_task_publish)
+        after_task_publish.connect(self.receiver_after_task_publish)
+
+    def connect_worker_signals(self):
+        from celery.signals import (
+            before_task_publish,
+            after_task_publish,
+            task_prerun,
+            task_retry,
+            task_success,
+            task_failure,
+            task_revoked,
+            task_unknown,
+            task_rejected,
+        )
 
-    before_task_publish.connect(receiver_before_task_publish)
-    after_task_publish.connect(receiver_after_task_publish)
+        before_task_publish.connect(self.receiver_before_task_publish)
+        after_task_publish.connect(self.receiver_after_task_publish)
+        task_prerun.connect(self.receiver_task_prerun)
+        task_retry.connect(self.receiver_task_retry)
+        task_success.connect(self.receiver_task_success)
+        task_failure.connect(self.receiver_task_failure)
+        task_revoked.connect(self.receiver_task_revoked)
+        task_unknown.connect(self.receiver_task_unknown)
+        task_rejected.connect(self.receiver_task_rejected)
diff --git a/django_structlog/celery/signals.py b/django_structlog/celery/signals.py
index 5374f41..e647924 100644
--- a/django_structlog/celery/signals.py
+++ b/django_structlog/celery/signals.py
@@ -22,12 +22,14 @@ modify_context_before_task_publish = django.dispatch.Signal()
 """ Signal to modify context passed over to ``celery`` task's context. You must modify the ``context`` dict.
 
 :param context: the context dict that will be passed over to the task runner's logger
+:param task_routing_key: routing key of the task
+:param task_properties: task's message properties
 
 >>> from django.dispatch import receiver
 >>> from django_structlog.celery import signals
 >>>
 >>> @receiver(signals.modify_context_before_task_publish)
-... def receiver_modify_context_before_task_publish(sender, signal, context, **kwargs):
+... def receiver_modify_context_before_task_publish(sender, signal, context, task_routing_key=None, task_properties=None, **kwargs):
 ...     keys_to_keep = {"request_id", "parent_task_id"}
 ...     new_dict = {
 ...         key_to_keep: context[key_to_keep]
diff --git a/django_structlog/celery/steps.py b/django_structlog/celery/steps.py
index 85914b5..9deb4a2 100644
--- a/django_structlog/celery/steps.py
+++ b/django_structlog/celery/steps.py
@@ -1,6 +1,6 @@
 from celery import bootsteps
 
-from . import receivers
+from .receivers import CeleryReceiver
 
 
 class DjangoStructLogInitStep(bootsteps.Step):
@@ -16,26 +16,5 @@ class DjangoStructLogInitStep(bootsteps.Step):
 
     def __init__(self, parent, **kwargs):
         super().__init__(parent, **kwargs)
-        import celery
-        from celery.signals import (
-            before_task_publish,
-            after_task_publish,
-            task_prerun,
-            task_retry,
-            task_success,
-            task_failure,
-            task_revoked,
-        )
-
-        before_task_publish.connect(receivers.receiver_before_task_publish)
-        after_task_publish.connect(receivers.receiver_after_task_publish)
-        task_prerun.connect(receivers.receiver_task_pre_run)
-        task_retry.connect(receivers.receiver_task_retry)
-        task_success.connect(receivers.receiver_task_success)
-        task_failure.connect(receivers.receiver_task_failure)
-        task_revoked.connect(receivers.receiver_task_revoked)
-        if celery.VERSION > (4,):
-            from celery.signals import task_unknown, task_rejected
-
-            task_unknown.connect(receivers.receiver_task_unknown)
-            task_rejected.connect(receivers.receiver_task_rejected)
+        self.receiver = CeleryReceiver()
+        self.receiver.connect_worker_signals()
diff --git a/django_structlog/commands.py b/django_structlog/commands.py
index 4a3096e..d1d1e96 100644
--- a/django_structlog/commands.py
+++ b/django_structlog/commands.py
@@ -2,39 +2,40 @@ import structlog
 import uuid
 
 logger = structlog.getLogger(__name__)
-stack = []
 
 
-def pre_receiver(sender, *args, **kwargs):
-    command_id = str(uuid.uuid4())
-    if len(stack):
-        parent_command_id, _ = stack[-1]
-        tokens = structlog.contextvars.bind_contextvars(
-            parent_command_id=parent_command_id, command_id=command_id
+class DjangoCommandReceiver:
+    def __init__(self):
+        self.stack = []
+
+    def pre_receiver(self, sender, *args, **kwargs):
+        command_id = str(uuid.uuid4())
+        if len(self.stack):
+            parent_command_id, _ = self.stack[-1]
+            tokens = structlog.contextvars.bind_contextvars(
+                parent_command_id=parent_command_id, command_id=command_id
+            )
+        else:
+            tokens = structlog.contextvars.bind_contextvars(command_id=command_id)
+        self.stack.append((command_id, tokens))
+
+        logger.info(
+            "command_started",
+            command_name=sender.__module__.replace(".management.commands", ""),
         )
-    else:
-        tokens = structlog.contextvars.bind_contextvars(command_id=command_id)
-    stack.append((command_id, tokens))
 
-    logger.info(
-        "command_started",
-        command_name=sender.__module__.replace(".management.commands", ""),
-    )
+    def post_receiver(self, sender, outcome, *args, **kwargs):
+        logger.info("command_finished")
 
+        if len(self.stack):  # pragma: no branch
+            command_id, tokens = self.stack.pop()
+            structlog.contextvars.reset_contextvars(**tokens)
 
-def post_receiver(sender, outcome, *args, **kwargs):
-    logger.info("command_finished")
+    def connect_signals(self):
+        try:
+            from django_extensions.management.signals import pre_command, post_command
+        except ModuleNotFoundError:  # pragma: no cover
+            return
 
-    if len(stack):
-        command_id, tokens = stack.pop()
-        structlog.contextvars.reset_contextvars(**tokens)
-
-
-def init_command_signals():
-    try:
-        from django_extensions.management.signals import pre_command, post_command
-    except ModuleNotFoundError:  # pragma: no cover
-        return
-
-    pre_command.connect(pre_receiver)
-    post_command.connect(post_receiver)
+        pre_command.connect(self.pre_receiver)
+        post_command.connect(self.post_receiver)
diff --git a/django_structlog_demo_project/home/views.py b/django_structlog_demo_project/home/views.py
index 448aabd..6ba3093 100644
--- a/django_structlog_demo_project/home/views.py
+++ b/django_structlog_demo_project/home/views.py
@@ -15,7 +15,7 @@ logger = structlog.get_logger(__name__)
 
 def enqueue_successful_task(request):
     logger.info("Enqueuing successful task")
-    successful_task.delay(foo="bar")
+    successful_task.apply_async(foo="bar", priority=5)
     return HttpResponse(status=201)
 
 
diff --git a/docs/celery.rst b/docs/celery.rst
index cd94a4f..8b08fa5 100644
--- a/docs/celery.rst
+++ b/docs/celery.rst
@@ -157,7 +157,7 @@ By example you can strip down the ``context`` to keep only some of the keys:
 .. code-block:: python
 
     @receiver(signals.modify_context_before_task_publish)
-    def receiver_modify_context_before_task_publish(sender, signal, context, **kwargs):
+    def receiver_modify_context_before_task_publish(sender, signal, context, task_routing_key=None, task_properties=None, **kwargs):
         keys_to_keep = {"request_id", "parent_task_id"}
         new_dict = {key_to_keep: context[key_to_keep] for key_to_keep in keys_to_keep if key_to_keep in context}
         context.clear()
diff --git a/docs/events.rst b/docs/events.rst
index 2909349..7ee08ff 100644
--- a/docs/events.rst
+++ b/docs/events.rst
@@ -122,6 +122,10 @@ These metadata appear once along with their associated event
 +------------------+------------------+----------------------------------------+
 | task_enqueued    | child_task_name  | name of the task being enqueued        |
 +------------------+------------------+----------------------------------------+
+| task_enqueued    | routing_key      | task's routing key                     |
++------------------+------------------+----------------------------------------+
+| task_enqueued    | priority         | priority of task (if any)              |
++------------------+------------------+----------------------------------------+
 | task_retrying    | reason           | reason for retry                       |
 +------------------+------------------+----------------------------------------+
 | task_started     | task             | name of the task                       |


Test patch:
diff --git a/test_app/tests/celery/test_receivers.py b/test_app/tests/celery/test_receivers.py
index 3d8dfb7..994513c 100644
--- a/test_app/tests/celery/test_receivers.py
+++ b/test_app/tests/celery/test_receivers.py
@@ -5,7 +5,7 @@ from unittest.mock import Mock, patch, call, MagicMock
 import structlog
 from celery import shared_task
 from django.contrib.auth.models import AnonymousUser
-from django.dispatch import receiver
+from django.dispatch import receiver as django_receiver
 from django.test import TestCase, RequestFactory
 
 from django_structlog.celery import receivers, signals
@@ -29,20 +29,13 @@ class TestReceivers(TestCase):
         def test_task(value):  # pragma: no cover
             pass
 
-        from celery.signals import before_task_publish, after_task_publish
-
-        before_task_publish.connect(receivers.receiver_before_task_publish)
-        after_task_publish.connect(receivers.receiver_after_task_publish)
-        try:
-            structlog.contextvars.bind_contextvars(request_id=expected_uuid)
-            with self.assertLogs(
-                logging.getLogger("django_structlog.celery.receivers"), logging.INFO
-            ) as log_results:
-                test_task.delay("foo")
-        finally:
-            before_task_publish.disconnect(receivers.receiver_before_task_publish)
-            after_task_publish.disconnect(receivers.receiver_after_task_publish)
-
+        receiver = receivers.CeleryReceiver()
+        receiver.connect_signals()
+        structlog.contextvars.bind_contextvars(request_id=expected_uuid)
+        with self.assertLogs(
+            logging.getLogger("django_structlog.celery.receivers"), logging.INFO
+        ) as log_results:
+            test_task.delay("foo")
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
         self.assertEqual("task_enqueued", record.msg["event"])
@@ -61,7 +54,8 @@ class TestReceivers(TestCase):
             user_id=expected_user_id,
             task_id=expected_parent_task_uuid,
         )
-        receivers.receiver_before_task_publish(headers=headers)
+        receiver = receivers.CeleryReceiver()
+        receiver.receiver_before_task_publish(headers=headers)
 
         self.assertDictEqual(
             {
@@ -90,8 +84,9 @@ class TestReceivers(TestCase):
         mock_conf = MagicMock()
         mock_current_app.conf = mock_conf
         mock_conf.task_protocol = 1
+        receiver = receivers.CeleryReceiver()
         with patch("celery.current_app", mock_current_app):
-            receivers.receiver_before_task_publish(headers=headers)
+            receiver.receiver_before_task_publish(headers=headers)
 
         self.assertDictEqual(
             {},
@@ -102,9 +97,16 @@ class TestReceivers(TestCase):
         expected_uuid = "00000000-0000-0000-0000-000000000000"
         user_id = "1234"
         expected_parent_task_uuid = "11111111-1111-1111-1111-111111111111"
+        routing_key = "foo"
+        properties = {"correlation_id": "22222222-2222-2222-2222-222222222222"}
+
+        received_properties = None
+        received_routing_key = None
 
-        @receiver(signals.modify_context_before_task_publish)
-        def receiver_modify_context_before_task_publish(sender, signal, context):
+        @django_receiver(signals.modify_context_before_task_publish)
+        def receiver_modify_context_before_task_publish(
+            sender, signal, context, task_properties, task_routing_key, **kwargs
+        ):
             keys_to_keep = {"request_id", "parent_task_id"}
             new_dict = {
                 key_to_keep: context[key_to_keep]
@@ -113,6 +115,10 @@ class TestReceivers(TestCase):
             }
             context.clear()
             context.update(new_dict)
+            nonlocal received_properties
+            received_properties = task_properties
+            nonlocal received_routing_key
+            received_routing_key = task_routing_key
 
         headers = {}
         structlog.contextvars.bind_contextvars(
@@ -120,7 +126,12 @@ class TestReceivers(TestCase):
             user_id=user_id,
             task_id=expected_parent_task_uuid,
         )
-        receivers.receiver_before_task_publish(headers=headers)
+        receiver = receivers.CeleryReceiver()
+        receiver.receiver_before_task_publish(
+            headers=headers,
+            routing_key=routing_key,
+            properties=properties,
+        )
 
         self.assertDictEqual(
             {
@@ -132,16 +143,21 @@ class TestReceivers(TestCase):
             headers,
             "Only `request_id` and `parent_task_id` are preserved",
         )
+        self.assertDictEqual(
+            {"correlation_id": "22222222-2222-2222-2222-222222222222"},
+            received_properties,
+        )
+        self.assertEqual("foo", received_routing_key)
 
     def test_receiver_after_task_publish(self):
         expected_task_id = "00000000-0000-0000-0000-000000000000"
         expected_task_name = "Foo"
         headers = {"id": expected_task_id, "task": expected_task_name}
-
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.INFO
         ) as log_results:
-            receivers.receiver_after_task_publish(headers=headers)
+            receiver.receiver_after_task_publish(headers=headers)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -152,15 +168,16 @@ class TestReceivers(TestCase):
         self.assertIn("child_task_name", record.msg)
         self.assertEqual(expected_task_name, record.msg["child_task_name"])
 
-    def test_receiver_after_task_publish_celery_3(self):
+    def test_receiver_after_task_publish_protocol_v1(self):
         expected_task_id = "00000000-0000-0000-0000-000000000000"
         expected_task_name = "Foo"
         body = {"id": expected_task_id, "task": expected_task_name}
 
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.INFO
         ) as log_results:
-            receivers.receiver_after_task_publish(body=body)
+            receiver.receiver_after_task_publish(body=body)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -186,11 +203,11 @@ class TestReceivers(TestCase):
 
         context = structlog.contextvars.get_merged_contextvars(self.logger)
         self.assertDictEqual({"foo": "bar"}, context)
-
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.INFO
         ) as log_results:
-            receivers.receiver_task_pre_run(task_id, task)
+            receiver.receiver_task_prerun(task_id, task)
         context = structlog.contextvars.get_merged_contextvars(self.logger)
 
         self.assertDictEqual(
@@ -210,7 +227,7 @@ class TestReceivers(TestCase):
         self.assertEqual("task_name", record.msg["task"])
 
     def test_signal_bind_extra_task_metadata(self):
-        @receiver(signals.bind_extra_task_metadata)
+        @django_receiver(signals.bind_extra_task_metadata)
         def receiver_bind_extra_request_metadata(
             sender, signal, task=None, logger=None
         ):
@@ -227,8 +244,8 @@ class TestReceivers(TestCase):
 
         context = structlog.contextvars.get_merged_contextvars(self.logger)
         self.assertDictEqual({"foo": "bar"}, context)
-
-        receivers.receiver_task_pre_run(task_id, task)
+        receiver = receivers.CeleryReceiver()
+        receiver.receiver_task_prerun(task_id, task)
         context = structlog.contextvars.get_merged_contextvars(self.logger)
 
         self.assertEqual(context["correlation_id"], expected_correlation_uuid)
@@ -237,10 +254,11 @@ class TestReceivers(TestCase):
     def test_receiver_task_retry(self):
         expected_reason = "foo"
 
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.WARNING
         ) as log_results:
-            receivers.receiver_task_retry(reason=expected_reason)
+            receiver.receiver_task_retry(reason=expected_reason)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -252,16 +270,17 @@ class TestReceivers(TestCase):
     def test_receiver_task_success(self):
         expected_result = "foo"
 
-        @receiver(signals.pre_task_succeeded)
+        @django_receiver(signals.pre_task_succeeded)
         def receiver_pre_task_succeeded(
             sender, signal, task=None, logger=None, result=None
         ):
             structlog.contextvars.bind_contextvars(result=result)
 
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.INFO
         ) as log_results:
-            receivers.receiver_task_success(result=expected_result)
+            receiver.receiver_task_success(result=expected_result)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -272,11 +291,11 @@ class TestReceivers(TestCase):
 
     def test_receiver_task_failure(self):
         expected_exception = "foo"
-
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.ERROR
         ) as log_results:
-            receivers.receiver_task_failure(exception=Exception("foo"))
+            receiver.receiver_task_failure(exception=Exception("foo"))
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -291,11 +310,11 @@ class TestReceivers(TestCase):
 
         mock_sender = Mock()
         mock_sender.throws = (Exception,)
-
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.INFO
         ) as log_results:
-            receivers.receiver_task_failure(
+            receiver.receiver_task_failure(
                 exception=Exception("foo"), sender=mock_sender
             )
 
@@ -319,10 +338,12 @@ class TestReceivers(TestCase):
         }
         request.task = expected_task_name
         request.id = task_id
+
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.WARNING
         ) as log_results:
-            receivers.receiver_task_revoked(
+            receiver.receiver_task_revoked(
                 request=request, terminated=False, signum=None, expired=False
             )
 
@@ -359,10 +380,12 @@ class TestReceivers(TestCase):
         }
         request.task = expected_task_name
         request.id = task_id
+
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.WARNING
         ) as log_results:
-            receivers.receiver_task_revoked(
+            receiver.receiver_task_revoked(
                 request=request, terminated=True, signum=SIGTERM, expired=False
             )
 
@@ -391,10 +414,11 @@ class TestReceivers(TestCase):
         task_id = "11111111-1111-1111-1111-111111111111"
         expected_task_name = "task_name"
 
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.ERROR
         ) as log_results:
-            receivers.receiver_task_unknown(id=task_id, name=expected_task_name)
+            receiver.receiver_task_unknown(id=task_id, name=expected_task_name)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -410,10 +434,11 @@ class TestReceivers(TestCase):
         message = Mock(name="message")
         message.properties = dict(correlation_id=task_id)
 
+        receiver = receivers.CeleryReceiver()
         with self.assertLogs(
             logging.getLogger("django_structlog.celery.receivers"), logging.ERROR
         ) as log_results:
-            receivers.receiver_task_rejected(message=message)
+            receiver.receiver_task_rejected(message=message)
 
         self.assertEqual(1, len(log_results.records))
         record = log_results.records[0]
@@ -422,23 +447,117 @@ class TestReceivers(TestCase):
         self.assertIn("task_id", record.msg)
         self.assertEqual(task_id, record.msg["task_id"])
 
+    def test_priority(self):
+        expected_uuid = "00000000-0000-0000-0000-000000000000"
+        user_id = "1234"
+        expected_parent_task_uuid = "11111111-1111-1111-1111-111111111111"
+        expected_routing_key = "foo"
+        expected_priority = 6
+        properties = {"priority": expected_priority}
+
+        headers = {}
+        structlog.contextvars.bind_contextvars(
+            request_id=expected_uuid,
+            user_id=user_id,
+            task_id=expected_parent_task_uuid,
+        )
+        receiver = receivers.CeleryReceiver()
+        receiver.receiver_before_task_publish(
+            headers=headers,
+            routing_key=expected_routing_key,
+            properties=properties,
+        )
+
+        self.assertDictEqual(
+            {
+                "__django_structlog__": {
+                    "user_id": user_id,
+                    "request_id": expected_uuid,
+                    "parent_task_id": expected_parent_task_uuid,
+                }
+            },
+            headers,
+            "Only `request_id` and `parent_task_id` are preserved",
+        )
+
+        expected_task_id = "00000000-0000-0000-0000-000000000000"
+        expected_task_name = "Foo"
+        headers = {"id": expected_task_id, "task": expected_task_name}
+
+        with self.assertLogs(
+            logging.getLogger("django_structlog.celery.receivers"), logging.INFO
+        ) as log_results:
+            receiver.receiver_after_task_publish(
+                headers=headers, routing_key=expected_routing_key
+            )
+
+        self.assertEqual(1, len(log_results.records))
+        record = log_results.records[0]
+        self.assertEqual("task_enqueued", record.msg["event"])
+        self.assertEqual("INFO", record.levelname)
+        self.assertIn("child_task_id", record.msg)
+        self.assertEqual(expected_task_id, record.msg["child_task_id"])
+        self.assertIn("child_task_name", record.msg)
+        self.assertEqual(expected_task_name, record.msg["child_task_name"])
+
+        self.assertIn("priority", record.msg)
+        self.assertEqual(expected_priority, record.msg["priority"])
+
+        self.assertIn("routing_key", record.msg)
+        self.assertEqual(expected_routing_key, record.msg["routing_key"])
+
+
+class TestConnectCeleryTaskSignals(TestCase):
+    def test_call(self):
+        from celery.signals import (
+            before_task_publish,
+            after_task_publish,
+            task_prerun,
+            task_retry,
+            task_success,
+            task_failure,
+            task_revoked,
+            task_unknown,
+            task_rejected,
+        )
+
+        from django_structlog.celery.receivers import CeleryReceiver
+
+        receiver = CeleryReceiver()
+        with patch(
+            "celery.utils.dispatch.signal.Signal.connect", autospec=True
+        ) as mock_connect:
+            receiver.connect_worker_signals()
+
+        mock_connect.assert_has_calls(
+            [
+                call(before_task_publish, receiver.receiver_before_task_publish),
+                call(after_task_publish, receiver.receiver_after_task_publish),
+                call(task_prerun, receiver.receiver_task_prerun),
+                call(task_retry, receiver.receiver_task_retry),
+                call(task_success, receiver.receiver_task_success),
+                call(task_failure, receiver.receiver_task_failure),
+                call(task_revoked, receiver.receiver_task_revoked),
+                call(task_unknown, receiver.receiver_task_unknown),
+                call(task_rejected, receiver.receiver_task_rejected),
+            ]
+        )
+
 
 class TestConnectCelerySignals(TestCase):
     def test_call(self):
         from celery.signals import before_task_publish, after_task_publish
-        from django_structlog.celery.receivers import (
-            receiver_before_task_publish,
-            receiver_after_task_publish,
-        )
+        from django_structlog.celery.receivers import CeleryReceiver
 
+        receiver = CeleryReceiver()
         with patch(
             "celery.utils.dispatch.signal.Signal.connect", autospec=True
         ) as mock_connect:
-            receivers.connect_celery_signals()
+            receiver.connect_signals()
 
         mock_connect.assert_has_calls(
             [
-                call(before_task_publish, receiver_before_task_publish),
-                call(after_task_publish, receiver_after_task_publish),
+                call(before_task_publish, receiver.receiver_before_task_publish),
+                call(after_task_publish, receiver.receiver_after_task_publish),
             ]
         )
diff --git a/test_app/tests/celery/test_steps.py b/test_app/tests/celery/test_steps.py
index 5cd3625..d323987 100644
--- a/test_app/tests/celery/test_steps.py
+++ b/test_app/tests/celery/test_steps.py
@@ -1,4 +1,4 @@
-from unittest.mock import patch, call
+from unittest.mock import patch
 
 from django.test import TestCase
 
@@ -7,44 +7,12 @@ from django_structlog.celery import steps
 
 class TestDjangoStructLogInitStep(TestCase):
     def test_call(self):
-        from celery.signals import (
-            before_task_publish,
-            after_task_publish,
-            task_prerun,
-            task_retry,
-            task_success,
-            task_failure,
-            task_revoked,
-            task_unknown,
-            task_rejected,
-        )
-        from django_structlog.celery.receivers import (
-            receiver_before_task_publish,
-            receiver_after_task_publish,
-            receiver_task_pre_run,
-            receiver_task_retry,
-            receiver_task_success,
-            receiver_task_failure,
-            receiver_task_revoked,
-            receiver_task_unknown,
-            receiver_task_rejected,
-        )
-
         with patch(
-            "celery.utils.dispatch.signal.Signal.connect", autospec=True
+            "django_structlog.celery.receivers.CeleryReceiver.connect_worker_signals",
+            autospec=True,
         ) as mock_connect:
-            steps.DjangoStructLogInitStep(None)
+            step = steps.DjangoStructLogInitStep(None)
+
+        mock_connect.assert_called_once()
 
-        mock_connect.assert_has_calls(
-            [
-                call(before_task_publish, receiver_before_task_publish),
-                call(after_task_publish, receiver_after_task_publish),
-                call(task_prerun, receiver_task_pre_run),
-                call(task_retry, receiver_task_retry),
-                call(task_success, receiver_task_success),
-                call(task_failure, receiver_task_failure),
-                call(task_revoked, receiver_task_revoked),
-                call(task_unknown, receiver_task_unknown),
-                call(task_rejected, receiver_task_rejected),
-            ]
-        )
+        self.assertIsNotNone(step.receiver)
diff --git a/test_app/tests/test_apps.py b/test_app/tests/test_apps.py
index 03c8e89..5efbe8b 100644
--- a/test_app/tests/test_apps.py
+++ b/test_app/tests/test_apps.py
@@ -1,8 +1,9 @@
-from unittest.mock import patch
+from unittest.mock import patch, create_autospec
 
 from django.test import TestCase
 
-from django_structlog import apps
+from django_structlog import apps, commands
+from django_structlog.celery import receivers
 
 
 class TestAppConfig(TestCase):
@@ -10,20 +11,62 @@ class TestAppConfig(TestCase):
         app = apps.DjangoStructLogConfig(
             "django_structlog", __import__("django_structlog")
         )
+        mock_receiver = create_autospec(spec=receivers.CeleryReceiver)
         with patch(
-            "django_structlog.celery.receivers.connect_celery_signals"
-        ) as mock_connect_celery_signals:
+            "django_structlog.celery.receivers.CeleryReceiver",
+            return_value=mock_receiver,
+        ):
             with self.settings(DJANGO_STRUCTLOG_CELERY_ENABLED=True):
                 app.ready()
-        mock_connect_celery_signals.assert_called_once()
+        mock_receiver.connect_signals.assert_called_once()
+
+        self.assertTrue(hasattr(app, "_celery_receiver"))
+        self.assertIsNotNone(app._celery_receiver)
 
     def test_celery_disabled(self):
         app = apps.DjangoStructLogConfig(
             "django_structlog", __import__("django_structlog")
         )
+
+        mock_receiver = create_autospec(spec=receivers.CeleryReceiver)
         with patch(
-            "django_structlog.celery.receivers.connect_celery_signals"
-        ) as mock_connect_celery_signals:
+            "django_structlog.celery.receivers.CeleryReceiver",
+            return_value=mock_receiver,
+        ):
             with self.settings(DJANGO_STRUCTLOG_CELERY_ENABLED=False):
                 app.ready()
-        mock_connect_celery_signals.assert_not_called()
+        mock_receiver.connect_signals.assert_not_called()
+
+        self.assertFalse(hasattr(app, "_celery_receiver"))
+
+    def test_command_enabled(self):
+        app = apps.DjangoStructLogConfig(
+            "django_structlog", __import__("django_structlog")
+        )
+        mock_receiver = create_autospec(spec=commands.DjangoCommandReceiver)
+        with patch(
+            "django_structlog.commands.DjangoCommandReceiver",
+            return_value=mock_receiver,
+        ):
+            with self.settings(DJANGO_STRUCTLOG_COMMAND_LOGGING_ENABLED=True):
+                app.ready()
+        mock_receiver.connect_signals.assert_called_once()
+
+        self.assertTrue(hasattr(app, "_django_command_receiver"))
+        self.assertIsNotNone(app._django_command_receiver)
+
+    def test_command_disabled(self):
+        app = apps.DjangoStructLogConfig(
+            "django_structlog", __import__("django_structlog")
+        )
+
+        mock_receiver = create_autospec(spec=commands.DjangoCommandReceiver)
+        with patch(
+            "django_structlog.commands.DjangoCommandReceiver",
+            return_value=mock_receiver,
+        ):
+            with self.settings(DJANGO_STRUCTLOG_COMMAND_LOGGING_ENABLED=False):
+                app.ready()
+        mock_receiver.connect_signals.assert_not_called()
+
+        self.assertFalse(hasattr(app, "_django_command_receiver"))