feat: adds request id

.github/dependabot.yml

@@ -7,6 +7,10 @@ updates:
       time: "08:00"
       day: "sunday"
     target-branch: "main"
+    groups:
+      patches:
+        update-types:
+          - "patch"
     reviewers:
       - "juntossomosmais/loyalty"
       - "juntossomosmais/loja-virtual"

.pre-commit-config.yaml

@@ -2,7 +2,7 @@
 # See https://pre-commit.com/hooks.html for more hooks
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.6.0
+    rev: v5.0.0
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer
@@ -29,7 +29,7 @@ repos:
             "--ignore-init-module-imports",
           ]
   - repo: https://github.com/psf/black
-    rev: 24.8.0
+    rev: 24.10.0
     hooks:
       - id: black
         exclude: migrations/
@@ -38,7 +38,7 @@ repos:
     hooks:
       - id: isort
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.11.2
+    rev: v1.14.1
     hooks:
       - id: mypy
         additional_dependencies:

django_outbox_pattern/consumers.py

@@ -2,11 +2,13 @@
 import logging
 
 from datetime import timedelta
+from uuid import uuid4
 
 from django import db
 from django.core.cache import cache
 from django.utils import timezone
 from django.utils.module_loading import import_string
+from request_id_django_log import local_threading
 from stomp.utils import get_uuid
 
 from django_outbox_pattern import settings
@@ -17,11 +19,25 @@
 
 
 def _get_msg_id(headers):
-    ret = None
-    for key, value in headers.items():
-        if key.endswith("-id"):
-            ret = value
-    return ret
+    """
+    Retrieves the first header value that matches either message-id, dop-msg-id or cap-msg-id.
+
+    These values are used to be added as the received message id, so we can track if the message was already received.
+
+    The cap-msg-id is a header that is used by the CAP .NET library, and it's used to identify the message.
+
+    The dop-msg-id is a header that is used by the Django Outbox Pattern library, and it's used to identify the message.
+    """
+    return headers.get("cap-msg-id") or headers.get("dop-msg-id") or headers.get("message-id")
+
+
+def _get_or_create_correlation_id(headers: dict) -> str:
+    if "dop-correlation-id" in headers:
+        return headers["dop-correlation-id"]
+
+    correlation_id = str(uuid4())
+    logger.debug("A new dop-correlation-id was generated %s", correlation_id)
+    return correlation_id
 
 
 class Consumer(Base):
@@ -38,6 +54,7 @@ def __init__(self, connection, username, passcode):
         self.set_listener(self.listener_name, self.listener_class(self))
 
     def message_handler(self, body, headers):
+        local_threading.request_id = _get_or_create_correlation_id(headers)
         try:
             body = json.loads(body)
         except json.JSONDecodeError as exc:
@@ -77,6 +94,7 @@ def message_handler(self, body, headers):
                 self._remove_old_messages()
             finally:
                 db.close_old_connections()
+            local_threading.request_id = None
 
     def start(self, callback, destination, queue_name=None):
         self.connect()

django_outbox_pattern/headers.py

@@ -1,23 +1,34 @@
 import json
 
+from uuid import uuid4
+
 from django.core.serializers.json import DjangoJSONEncoder
 from django.utils import timezone
 from django.utils.module_loading import import_string
+from request_id_django_log.request_id import current_request_id
+from request_id_django_log.settings import NO_REQUEST_ID
 
 from django_outbox_pattern import settings
 
 
 def generate_headers(message):
+    correlation_id = current_request_id()
+    if not correlation_id or correlation_id == NO_REQUEST_ID:
+        correlation_id = uuid4()
+
     return {
         "dop-msg-id": str(message.id),
         "dop-msg-destination": message.destination,
         "dop-msg-type": message.__class__.__name__,
         "dop-msg-sent-time": timezone.now(),
+        "dop-correlation-id": correlation_id,
     }
 
 
 def get_message_headers(published):
-    if not published.headers:
-        default_headers = import_string(settings.DEFAULT_GENERATE_HEADERS)
-        return json.loads(json.dumps(default_headers(published), cls=DjangoJSONEncoder))
-    return published.headers
+    default_headers = import_string(settings.DEFAULT_GENERATE_HEADERS)(published)
+    return json.loads(
+        json.dumps(
+            default_headers if not published.headers else published.headers | default_headers, cls=DjangoJSONEncoder
+        )
+    )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-outbox-pattern"
-version = "2.1.0"
+version = "3.0.0"
 description = "A django application to make it easier to use the transactional outbox pattern"
 license = "MIT"
 authors = ["Hugo Brilhante <hugobrilhante@gmail.com>"]
@@ -32,11 +32,12 @@ packages = [
 python = ">=3.10,<4.0"
 django = ">=5.0.8"
 "stomp.py" = ">=8.0.1,<9"
+request-id-django-log = "^0.2.0"
 
 [tool.poetry.group.dev.dependencies]
 psycopg2-binary = "^2.9.6"
 coverage = "*"
-pre-commit = "^3.8.0"
+pre-commit = "*"
 
 [tool.black]
 line-length = 120

tests/integration/test_consumer.py

@@ -47,7 +47,7 @@ def test_when_send_message_twice(self):
         callback = get_callback()
         destination = "/topic/consumer.v1"
         body = '{"message": "Message twice"}'
-        headers = {"msg-id": "fbb5aaf7-8c0b-453e-a23c-1b8a072a2573"}
+        headers = {"dop-msg-id": "fbb5aaf7-8c0b-453e-a23c-1b8a072a2573"}
         self.consumer.start(callback, destination)
         self.consumer.connection.send(destination=destination, body=body, headers=headers)
         self.consumer.connection.send(destination=destination, body=body, headers=headers)

tests/unit/test_consumer.py

@@ -1,11 +1,15 @@
 from unittest.mock import Mock
 from unittest.mock import patch
+from uuid import uuid4
 
 from django.db import transaction
+from django.test import SimpleTestCase
 from django.test import TransactionTestCase
+from request_id_django_log import local_threading
 from stomp.exception import StompException
 
 from django_outbox_pattern.choices import StatusChoice
+from django_outbox_pattern.consumers import _get_or_create_correlation_id
 from django_outbox_pattern.factories import factory_consumer
 from django_outbox_pattern.payloads import Payload
 
@@ -103,3 +107,32 @@ def test_consumer_start_with_correct_headers(self):
         self.assertIn("x-queue-name", self.consumer.subscribe_headers)
         self.assertIn("x-dead-letter-routing-key", self.consumer.subscribe_headers)
         self.assertIn("x-dead-letter-exchange", self.consumer.subscribe_headers)
+
+    def test_consumer_message_handler_should_add_correlation_id_from_header_into_local_threading(self):
+        self.consumer.callback = lambda p: p.save()
+
+        self.consumer.message_handler('{"message": "my message"}', {"message-id": 1, "dop-correlation-id": "1234"})
+
+        self.assertEqual(self.consumer.received_class.objects.filter(status=StatusChoice.SUCCEEDED).count(), 1)
+        message = self.consumer.received_class.objects.filter(status=StatusChoice.SUCCEEDED).first()
+        self.assertEqual({"message": "my message"}, message.body)
+        self.assertEqual({"message-id": 1, "dop-correlation-id": "1234"}, message.headers)
+        self.assertEqual("1", message.msg_id)
+        self.assertIsNone(local_threading.request_id)
+
+
+class GetOrCreateCorrelationIdTest(SimpleTestCase):
+
+    def test_should_return_correlation_id_from_headers(self):
+        headers = {"dop-correlation-id": "1234"}
+        with patch(f"{_get_or_create_correlation_id.__module__}.uuid4", wraps=uuid4) as uuid4_spy:
+            self.assertEqual("1234", _get_or_create_correlation_id(headers))
+            uuid4_spy.assert_not_called()
+
+        self.assertEqual("1234", _get_or_create_correlation_id(headers))
+
+    def test_should_create_a_new_correlation_id_given_header_without_correlation_id(self) -> None:
+        headers: dict = {}
+        with patch(f"{_get_or_create_correlation_id.__module__}.uuid4", wraps=uuid4) as uuid4_spy:
+            self.assertIsNotNone(_get_or_create_correlation_id(headers))
+            uuid4_spy.assert_called_once()

tests/unit/test_models.py

@@ -1,5 +1,9 @@
+from uuid import uuid4
+
 from django.test import TransactionTestCase
+from request_id_django_log import local_threading
 
+from django_outbox_pattern.models import Published
 from django_outbox_pattern.models import Received
 
 
@@ -15,3 +19,20 @@ def test_received_should_return_destination_value_when_have_headers(self):
     def test_received_should_return_destination_empty_when_object_have_headers_but_without_key_destination(self):
         received = Received(headers={"fake": "fake"})
         self.assertEqual("", received.destination)
+
+
+class PublishedTest(TransactionTestCase):
+
+    def test_published_should_add_correlation_id_header_from_current_request_id(self):
+        request_id = str(uuid4())
+        local_threading.request_id = request_id
+        published = Published.objects.create(destination="destination", body={"message": "Message test"})
+        self.assertEqual(published.headers["dop-correlation-id"], request_id)
+
+    def test_published_should_add_correlation_id_given_custom_header(self):
+        request_id = str(uuid4())
+        local_threading.request_id = request_id
+        published = Published.objects.create(
+            destination="destination", body={"message": "Message test"}, headers={"custom": "xpto-lalala"}
+        )
+        self.assertIn("dop-correlation-id", published.headers)

tests/unit/test_producer.py

@@ -1,7 +1,9 @@
 from unittest.mock import Mock
 from unittest.mock import patch
+from uuid import uuid4
 
 from django.test import TestCase
+from request_id_django_log import local_threading
 from stomp.exception import StompException
 
 from django_outbox_pattern import settings
@@ -23,6 +25,17 @@ def test_producer_send(self):
         self.assertEqual(self.producer.connection.send.call_count, 1)
         self.assertTrue(published.headers is not None)
 
+    def test_producer_send_should_add_correlation_id_header_from_current_request_id(self):
+        request_id = str(uuid4())
+        local_threading.request_id = request_id
+        published = Published.objects.create(destination="destination", body={"message": "Message test"})
+        self.producer.start()
+        self.producer.send(published)
+        self.producer.stop()
+        self.assertEqual(self.producer.connection.send.call_count, 1)
+        self.assertIsNotNone(published.headers)
+        self.assertEqual(published.headers["dop-correlation-id"], request_id)
+
     def test_producer_send_with_header(self):
         headers = {"key": "value"}
         published = Published.objects.create(
@@ -33,7 +46,7 @@ def test_producer_send_with_header(self):
         self.producer.stop()
         self.assertEqual(self.producer.connection.send.call_count, 1)
         self.assertTrue(published.headers is not None)
-        self.assertEqual(published.headers, headers)
+        self.assertEqual(published.headers["key"], headers["key"])
 
     def test_producer_send_event(self):
         self.producer.start()