make lint happy

Signed-off-by: Roberto Rodriguez <[email protected]>

Author: Cyb3rWard0g

dapr_agents/workflow/decorators/messaging.py

@@ -16,6 +16,7 @@
     "Union types, forward references, and explicit Dapr workflow integration."
 )
 
+
 def message_router(
     func: Optional[Callable[..., Any]] = None,
     *,
@@ -46,13 +47,12 @@ def message_router(
     """
 
     def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
-
         warnings.warn(
             _MESSAGE_ROUTER_DEPRECATION_MESSAGE,
             DeprecationWarning,
             stacklevel=2,
         )
-        
+
         is_workflow = hasattr(f, "_is_workflow")
         workflow_name = getattr(f, "_workflow_name", None)
 
@@ -74,7 +74,8 @@ def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
 
         logger.debug(
             "@message_router (legacy): '%s' => models %s",
-            f.__name__, [m.__name__ for m in message_models],
+            f.__name__,
+            [m.__name__ for m in message_models],
         )
 
         # Attach metadata for later registration

dapr_agents/workflow/decorators/routers.py

@@ -50,7 +50,9 @@ def on_event(message: Union[Foo, Bar]): ...
     def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
         # Validate required kwargs only when decorator is used with args
         if pubsub is None or topic is None:
-            raise ValueError("`pubsub` and `topic` are required when using @message_router with arguments.")
+            raise ValueError(
+                "`pubsub` and `topic` are required when using @message_router with arguments."
+            )
 
         sig = inspect.signature(f)
         if "message" not in sig.parameters:
@@ -60,7 +62,9 @@ def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
         try:
             hints = get_type_hints(f, globalns=f.__globals__)
         except Exception:
-            logger.debug("Failed to fully resolve type hints for %s", f.__name__, exc_info=True)
+            logger.debug(
+                "Failed to fully resolve type hints for %s", f.__name__, exc_info=True
+            )
             hints = getattr(f, "__annotations__", {}) or {}
 
         raw_hint = hints.get("message")
@@ -72,7 +76,9 @@ def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
 
         models = extract_message_models(raw_hint)
         if not models:
-            raise TypeError(f"Unsupported or unresolved message type for '{f.__name__}': {raw_hint!r}")
+            raise TypeError(
+                f"Unsupported or unresolved message type for '{f.__name__}': {raw_hint!r}"
+            )
 
         # Optional early validation of supported schema kinds
         for m in models:
@@ -82,10 +88,11 @@ def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
         data = {
             "pubsub": pubsub,
             "topic": topic,
-            "dead_letter_topic": dead_letter_topic or (f"{topic}_DEAD" if topic else None),
+            "dead_letter_topic": dead_letter_topic
+            or (f"{topic}_DEAD" if topic else None),
             "is_broadcast": broadcast,
-            "message_schemas": models,                     # list[type]
-            "message_types": [m.__name__ for m in models], # list[str]
+            "message_schemas": models,  # list[type]
+            "message_types": [m.__name__ for m in models],  # list[str]
         }
 
         # Attach metadata; deepcopy for defensive isolation
@@ -94,9 +101,13 @@ def decorator(f: Callable[..., Any]) -> Callable[..., Any]:
 
         logger.debug(
             "@message_router: '%s' => models %s (topic=%s, pubsub=%s, broadcast=%s)",
-            f.__name__, [m.__name__ for m in models], topic, pubsub, broadcast
+            f.__name__,
+            [m.__name__ for m in models],
+            topic,
+            pubsub,
+            broadcast,
         )
         return f
 
     # Support both @message_router(...) and bare @message_router usage
-    return decorator if func is None else decorator(func)
+    return decorator if func is None else decorator(func)

dapr_agents/workflow/utils/registration.py

@@ -73,7 +73,9 @@ def _iter_handlers(obj: Any):
             schemas: List[Type[Any]] = meta.get("message_schemas") or []
 
             # Bind method to instance if needed (descriptor protocol).
-            bound = handler if owner is None else handler.__get__(owner, owner.__class__)
+            bound = (
+                handler if owner is None else handler.__get__(owner, owner.__class__)
+            )
 
             async def _invoke(
                 bound_handler: Callable[..., Any],
@@ -92,20 +94,25 @@ def _make_handler(
                 bound_handler: Callable[..., Any],
             ) -> Callable[[SubscriptionMessage], TopicEventResponse]:
                 """Create a Dapr-compatible handler for a single decorated function."""
+
                 def handler_fn(message: SubscriptionMessage) -> TopicEventResponse:
                     try:
                         # 1) Extract payload + CloudEvent metadata (bytes/str/dict are also supported by the extractor)
                         event_data, metadata = extract_cloudevent_data(message)
 
                         # 2) Validate against the first matching schema (or dict as fallback)
                         parsed = None
-                        for model in (schemas or [dict]):
+                        for model in schemas or [dict]:
                             try:
                                 parsed = validate_message_model(model, event_data)
                                 break
                             except Exception:
                                 # Try the next schema; log at debug for signal without noise.
-                                logger.debug("Schema %r did not match payload; trying next.", model, exc_info=True)
+                                logger.debug(
+                                    "Schema %r did not match payload; trying next.",
+                                    model,
+                                    exc_info=True,
+                                )
                                 continue
 
                         if parsed is None:
@@ -125,7 +132,9 @@ def handler_fn(message: SubscriptionMessage) -> TopicEventResponse:
 
                         # 4) Bridge worker thread → event loop
                         if loop and loop.is_running():
-                            fut = asyncio.run_coroutine_threadsafe(_invoke(bound_handler, parsed), loop)
+                            fut = asyncio.run_coroutine_threadsafe(
+                                _invoke(bound_handler, parsed), loop
+                            )
                             return fut.result()
                         return asyncio.run(_invoke(bound_handler, parsed))
 
@@ -144,4 +153,4 @@ def handler_fn(message: SubscriptionMessage) -> TopicEventResponse:
             )
             closers.append(close_fn)
 
-    return closers
+    return closers

dapr_agents/workflow/utils/routers.py

@@ -34,8 +34,7 @@ def extract_message_models(type_hint: Any) -> list[type]:
     origin = get_origin(type_hint)
     if origin in (Union, types.UnionType):  # handle both `Union[...]` and `A | B`
         return [
-            t for t in get_args(type_hint)
-            if t is not NoneType and isinstance(t, type)
+            t for t in get_args(type_hint) if t is not NoneType and isinstance(t, type)
         ]
 
     return [type_hint] if isinstance(type_hint, type) else []
@@ -154,7 +153,9 @@ def extract_cloudevent_data(
         raise ValueError(f"Unexpected message type: {type(message)!r}")
 
     if not isinstance(event_data, dict):
-        logger.debug("Event data is not a dict (type=%s); value=%r", type(event_data), event_data)
+        logger.debug(
+            "Event data is not a dict (type=%s); value=%r", type(event_data), event_data
+        )
 
     return event_data, metadata
 

quickstarts/04-message-router-workflow/app.py

@@ -43,7 +43,7 @@ def _set_stop(*_: object) -> None:
 
 async def main() -> None:
     runtime = wf.WorkflowRuntime()
-    
+
     runtime.register_workflow(blog_workflow)
     runtime.register_activity(create_outline)
     runtime.register_activity(write_post)
@@ -74,4 +74,4 @@ async def main() -> None:
     try:
         asyncio.run(main())
     except KeyboardInterrupt:
-        pass
+        pass

quickstarts/04-message-router-workflow/handlers.py

@@ -30,8 +30,10 @@ def start_blog_workflow(message: StartBlogMessage) -> TopicEventResponse:
             workflow=blog_workflow,
             input=message.model_dump(),
         )
-        logger.info("Scheduled blog_workflow instance=%s topic=%s", instance_id, message.topic)
+        logger.info(
+            "Scheduled blog_workflow instance=%s topic=%s", instance_id, message.topic
+        )
         return TopicEventResponse("success")
     except Exception as exc:  # transient infra error → retry
         logger.exception("Failed to schedule blog workflow: %s", exc)
-        return TopicEventResponse("retry")
+        return TopicEventResponse("retry")

quickstarts/04-message-router-workflow/message_client.py

@@ -152,4 +152,4 @@ def _stop(*_: object) -> None:
 
 
 if __name__ == "__main__":
-    sys.exit(asyncio.run(main()))
+    sys.exit(asyncio.run(main()))