[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

src/workflows/recipe/__init__.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
-from contextlib import ExitStack
+
 import functools
 import logging
 from collections.abc import Callable
+from contextlib import ExitStack
 from typing import Any
 
 from opentelemetry import trace
@@ -80,8 +81,6 @@ def unwrap_recipe(header, message):
             if recipe_id:
                 span.set_attribute("recipe_id", recipe_id)
 
-            
-
             # Extract span_id and trace_id for logging
             span_context = span.get_span_context()
             if span_context and span_context.is_valid:
@@ -92,21 +91,29 @@ def unwrap_recipe(header, message):
                     "span_id": span_id,
                     "trace_id": trace_id,
                 }
-                
+
                 if recipe_id:
                     otel_logs["recipe_id"] = recipe_id
-        
+
             with ExitStack() as stack:
                 # Configure the context depending on if service is emitting spans
-                if otel_logs and log_extender and rw.environment and rw.environment.get("ID"):
-                    stack.enter_context(log_extender('recipe_ID', rw.environment.get("ID")))
-                    stack.enter_context(log_extender('otel_logs', otel_logs))
+                if (
+                    otel_logs
+                    and log_extender
+                    and rw.environment
+                    and rw.environment.get("ID")
+                ):
+                    stack.enter_context(
+                        log_extender("recipe_ID", rw.environment.get("ID"))
+                    )
+                    stack.enter_context(log_extender("otel_logs", otel_logs))
                 elif log_extender and rw.environment and rw.environment.get("ID"):
-                    stack.enter_context(log_extender('recipe_ID', rw.environment.get("ID")))
+                    stack.enter_context(
+                        log_extender("recipe_ID", rw.environment.get("ID"))
+                    )
 
                 return callback(rw, header, message.get("payload"))
-    
-        
+
         if allow_non_recipe_messages:
             return callback(None, header, message)
         #   self.log.warning('Discarding non-recipe message:\n' + \

src/workflows/services/common_service.py

@@ -192,10 +192,12 @@ def start_transport(self):
             self.transport.subscription_callback_set_intercept(
                 self._transport_interceptor
             )
-            
+
             # Configure OTELTracing if configuration is available
             otel_config = (
-                self.config._opentelemetry if self.config and hasattr(self.config, "opentelemetry") else None
+                self.config._opentelemetry
+                if self.config and hasattr(self.config, "opentelemetry")
+                else None
             )
             if otel_config:
                 # Configure OTELTracing

src/workflows/transport/common_transport.py

@@ -10,6 +10,7 @@
 
 MessageCallback = Callable[[Mapping[str, Any], Any], None]
 
+
 class TemporarySubscription(NamedTuple):
     subscription_id: int
     queue_name: str

src/workflows/transport/middleware/__init__.py

@@ -236,7 +236,7 @@ def wrap(f: Callable):
     # debugging
     if f.__name__ == "send":
         print("we are wrapping send now")
-    
+
     @functools.wraps(f)
     def wrapper(self, *args, **kwargs):
         return functools.reduce(
@@ -248,4 +248,4 @@ def wrapper(self, *args, **kwargs):
         )(*args, **kwargs)
 
     print(wrapper.__wrapped__)
-    return wrapper
+    return wrapper

src/workflows/transport/middleware/otel_tracing.py

@@ -1,15 +1,15 @@
 from __future__ import annotations
 
 import functools
+import json
 from collections.abc import Callable
 
 from opentelemetry import trace
-from opentelemetry.propagate import extract, inject
 from opentelemetry.context import Context
+from opentelemetry.propagate import extract, inject
+
+from workflows.transport.common_transport import MessageCallback, TemporarySubscription
 
-from workflows.transport.middleware import BaseTransportMiddleware
-from workflows.transport.common_transport import TemporarySubscription, MessageCallback
-import json
 
 class OTELTracingMiddleware:
     def __init__(self, tracer: trace.Tracer, service_name: str):
@@ -19,7 +19,9 @@ def __init__(self, tracer: trace.Tracer, service_name: str):
     def send(self, call_next: Callable, destination: str, message: Any, **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
+        parent_context = (
+            trace.set_span_in_context(current_span) if current_span else None
+        )
 
         with self.tracer.start_as_current_span(
             "transport.send",
@@ -29,8 +31,7 @@ def send(self, call_next: Callable, destination: str, message: Any, **kwargs):
 
             span.set_attribute("message", json.dumps(message))
             span.set_attribute("destination", destination)
-            print("parent_context is...",parent_context)
-
+            print("parent_context is...", parent_context)
 
             # Inject the current trace context into the message headers
             headers = kwargs.get("headers", {})
@@ -41,7 +42,9 @@ def send(self, call_next: Callable, destination: str, message: Any, **kwargs):
 
             return call_next(destination, message, **kwargs)
 
-    def subscribe(self, call_next: Callable, channel: str, callback: Callable, **kwargs) -> int:
+    def subscribe(
+        self, call_next: Callable, channel: str, callback: Callable, **kwargs
+    ) -> int:
         @functools.wraps(callback)
         def wrapped_callback(header, message):
             # Extract trace context from message headers
@@ -63,13 +66,15 @@ def wrapped_callback(header, message):
 
         return call_next(channel, wrapped_callback, **kwargs)
 
-    def subscribe_broadcast(self, call_next: Callable, channel: str, callback: Callable, **kwargs) -> int:
+    def subscribe_broadcast(
+        self, call_next: Callable, channel: str, callback: Callable, **kwargs
+    ) -> int:
         @functools.wraps(callback)
         def wrapped_callback(header, message):
             # Extract trace context from message headers
             ctx = extract(header) if header else Context()
 
-    #         # Start a new span with the extracted context
+            #         # Start a new span with the extracted context
             with self.tracer.start_as_current_span(
                 "transport.subscribe_broadcast",
                 context=ctx,
@@ -119,7 +124,9 @@ def unsubscribe(
     ):
         # Get current span context
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transport.unsubscribe",
@@ -141,7 +148,9 @@ def ack(
     ):
         # Get current span context
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transport.ack",
@@ -163,7 +172,9 @@ def nack(
     ):
         # Get current span context
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transport.nack",
@@ -183,7 +194,9 @@ def transaction_begin(
         """Start a new transaction span"""
         # Get current span context (may be None if this is the root span)
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transaction.begin",
@@ -196,11 +209,15 @@ def transaction_begin(
 
             return call_next(subscription_id=subscription_id, **kwargs)
 
-    def transaction_abort(self, call_next: Callable, transaction_id: int | None = None, **kwargs):
+    def transaction_abort(
+        self, call_next: Callable, transaction_id: int | None = None, **kwargs
+    ):
         """Abort a transaction span"""
         # Get current span context
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transaction.abort",
@@ -213,11 +230,15 @@ def transaction_abort(self, call_next: Callable, transaction_id: int | None = No
 
             call_next(transaction_id=transaction_id, **kwargs)
 
-    def transaction_commit(self, call_next: Callable, transaction_id: int | None = None, **kwargs):
+    def transaction_commit(
+        self, call_next: Callable, transaction_id: int | None = None, **kwargs
+    ):
         """Commit a transaction span"""
         # Get current span context
         current_span = trace.get_current_span()
-        current_context = trace.set_span_in_context(current_span) if current_span else Context()
+        current_context = (
+            trace.set_span_in_context(current_span) if current_span else Context()
+        )
 
         with self.tracer.start_as_current_span(
             "transaction.commit",
@@ -228,4 +249,3 @@ def transaction_commit(self, call_next: Callable, transaction_id: int | None = N
                 span.set_attribute("transaction_id", transaction_id)
 
             call_next(transaction_id=transaction_id, **kwargs)
-