Merge branch 'main' into renovate/major-all

Author: ndevenish

.github/workflows/python.yml

@@ -62,4 +62,4 @@ jobs:
           name: artifact
           path: dist
       - name: Publish package distributions to PyPI
-        uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0
+        uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # v1.14.0

.pre-commit-config.yaml

@@ -16,7 +16,7 @@ repos:
   - id: check-yaml
 
 - repo: https://github.com/astral-sh/ruff-pre-commit
-  rev: v0.14.10
+  rev: v0.15.9
   hooks:
     - id: ruff
       args: [--fix, --exit-non-zero-on-fix, --show-fixes]
@@ -31,7 +31,7 @@ repos:
 
 # Type checking
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.19.1
+  rev: v1.20.0
   hooks:
   - id: mypy
     files: 'src/.*\.py$'

requirements_dev.txt

@@ -1,7 +1,7 @@
 bidict==0.23.1
 pika==1.3.2
 prometheus_client==0.21.0
-pytest==8.3.3
+pytest==9.0.3
 pytest-cov==7.1.0
 pytest-mock==3.14.0
 pytest-timeout==2.3.1

src/workflows/transport/middleware/otel_tracing.py

@@ -107,6 +107,72 @@ def wrapped_callback(header, message):
 
         return call_next(channel_hint, wrapped_callback, **kwargs)
 
+    def raw_send(self, call_next: Callable, destination: str, message, **kwargs):
+        # Get current span context (may be None if this is the root span)
+        current_span = trace.get_current_span()
+        parent_context = (
+            trace.set_span_in_context(current_span) if current_span else None
+        )
+
+        with self.tracer.start_as_current_span(
+            "transport.raw_send",
+            context=parent_context,
+        ) as span:
+            self._set_span_attributes(span, destination=destination)
+
+            # Inject the current trace context into the message headers
+            headers = kwargs.get("headers", {})
+            if headers is None:
+                headers = {}
+            inject(headers)  # This modifies headers in-place
+            kwargs["headers"] = headers
+
+            return call_next(destination, message, **kwargs)
+
+    def broadcast(self, call_next: Callable, destination: str, message, **kwargs):
+        # Get current span context (may be None if this is the root span)
+        current_span = trace.get_current_span()
+        parent_context = (
+            trace.set_span_in_context(current_span) if current_span else None
+        )
+
+        with self.tracer.start_as_current_span(
+            "transport.broadcast",
+            context=parent_context,
+        ) as span:
+            self._set_span_attributes(span, destination=destination)
+
+            # Inject the current trace context into the message headers
+            headers = kwargs.get("headers", {})
+            if headers is None:
+                headers = {}
+            inject(headers)  # This modifies headers in-place
+            kwargs["headers"] = headers
+
+            return call_next(destination, message, **kwargs)
+
+    def raw_broadcast(self, call_next: Callable, destination: str, message, **kwargs):
+        # Get current span context (may be None if this is the root span)
+        current_span = trace.get_current_span()
+        parent_context = (
+            trace.set_span_in_context(current_span) if current_span else None
+        )
+
+        with self.tracer.start_as_current_span(
+            "transport.raw_broadcast",
+            context=parent_context,
+        ) as span:
+            self._set_span_attributes(span, destination=destination)
+
+            # Inject the current trace context into the message headers
+            headers = kwargs.get("headers", {})
+            if headers is None:
+                headers = {}
+            inject(headers)  # This modifies headers in-place
+            kwargs["headers"] = headers
+
+            return call_next(destination, message, **kwargs)
+
     def unsubscribe(
         self,
         call_next: Callable,