experiment

Author: GefMar

docs/examples/router/multiple_brokers.py

@@ -6,9 +6,9 @@
 
 router = TaskiqRouter()
 
-default_email_flow = Flow.queue("emails.default")
-priority_email_flow = Flow.queue("emails.priority")
-bulk_email_flow = Flow.queue("emails.bulk")
+default_email_flow = Flow("emails.default")
+priority_email_flow = Flow("emails.priority")
+bulk_email_flow = Flow("emails.bulk")
 
 default_broker = InMemoryBroker(
     router=router,
@@ -30,9 +30,9 @@ async def send_email(user_id: int, template: str) -> str:
     return f"{template} email sent to user {user_id}"
 
 
-router.route_task(
-    send_email.task_name,
-    broker="priority",
+priority_route = router.route_task(
+    send_email,
+    broker=priority_broker,
     flow=priority_email_flow,
 )
 
@@ -46,14 +46,18 @@ async def _main() -> None:
         routed_task = await send_email.kiq(7, "welcome")
         routed_result = await routed_task.wait_result(timeout=2)
 
-        bulk_task = await send_email.kicker().with_route(
-            "default",
-            bulk_email_flow,
-        ).kiq(8, "digest")
+        bulk_task = (
+            await send_email.kicker()
+            .with_route(
+                default_broker,
+                bulk_email_flow,
+            )
+            .kiq(8, "digest")
+        )
         bulk_result = await bulk_task.wait_result(timeout=2)
 
         print(f"Direct call: {direct_result}")
-        print(f"Default route: {router.resolve_route(send_email.task_name)}")
+        print(f"Declared route: {priority_route.broker_name}")
         print(f"Routed call: {routed_result.return_value}")
         print(f"Route override: {bulk_result.return_value}")
     finally:

docs/examples/router/shared_task_package.py

@@ -1,19 +1,36 @@
 """Declare shared task definitions and bind them in the final application."""
 
 import asyncio
+from collections.abc import Mapping
+from dataclasses import dataclass
 
 from taskiq import Flow, InMemoryBroker, TaskiqRouter, task_builder
 
 
+@dataclass(frozen=True, slots=True)
+class BillingQueue:
+    """Broker-specific flow that follows the shared flow protocol."""
+
+    name: str
+    priority: int
+
+    def broker_options(self, broker_name: str) -> Mapping[str, object]:
+        """Return options that a billing broker adapter can understand."""
+        return {
+            "broker": broker_name,
+            "priority": self.priority,
+        }
+
+
 @task_builder("billing.calculate_total", domain="billing")
 async def calculate_total(price: int, quantity: int) -> int:
     """Package-level task definition that is not bound to any broker."""
     return price * quantity
 
 
 router = TaskiqRouter()
-billing_flow = Flow.queue("billing.tasks")
-priority_billing_flow = Flow.queue("billing.priority")
+billing_flow = Flow("billing.tasks")
+priority_billing_flow = BillingQueue(name="billing.priority", priority=10)
 
 billing_broker = InMemoryBroker(
     router=router,
@@ -22,17 +39,26 @@ async def calculate_total(price: int, quantity: int) -> int:
     await_inplace=True,
 )
 
-registered_calculate_total = billing_broker.register_task(calculate_total)
+registered_calculate_total = router.register_task(
+    calculate_total,
+    broker=billing_broker,
+    flow=billing_flow,
+)
 
 
 async def _main() -> None:
     await billing_broker.startup()
     try:
         direct_result = await calculate_total.call(19, 3)
 
-        prepared_task = registered_calculate_total.kicker().with_flow(
-            priority_billing_flow,
-        ).prepare(19, 3)
+        prepared_task = (
+            registered_calculate_total.kicker()
+            .with_route(
+                billing_broker,
+                priority_billing_flow,
+            )
+            .prepare(19, 3)
+        )
 
         queued_task = await prepared_task.kiq()
         queued_result = await queued_task.wait_result(timeout=2)

taskiq/__init__.py

@@ -24,7 +24,7 @@
     TaskiqError,
     TaskiqResultTimeoutError,
 )
-from taskiq.flow import Flow, FlowKind
+from taskiq.flow import Flow, FlowProtocol
 from taskiq.funcs import gather
 from taskiq.kicker import PreparedKiq
 from taskiq.message import BrokerMessage, TaskiqMessage
@@ -52,7 +52,7 @@
     "BrokerMessage",
     "Context",
     "Flow",
-    "FlowKind",
+    "FlowProtocol",
     "InMemoryBroker",
     "NoResultError",
     "PreparedKiq",

taskiq/abc/broker.py

@@ -24,7 +24,7 @@
 from taskiq.decor import AsyncTaskiqDecoratedTask
 from taskiq.events import TaskiqEvents
 from taskiq.exceptions import TaskBrokerMismatchError
-from taskiq.flow import Flow
+from taskiq.flow import FlowProtocol
 from taskiq.formatters.proxy_formatter import ProxyFormatter
 from taskiq.message import BrokerMessage
 from taskiq.result_backends.dummy import DummyResultBackend
@@ -84,7 +84,7 @@ def __init__(
         *,
         router: TaskiqRouter | None = None,
         broker_name: str | None = None,
-        default_flow: Flow | None = None,
+        default_flow: FlowProtocol | None = None,
     ) -> None:
         if result_backend is None:
             result_backend = DummyResultBackend()
@@ -262,7 +262,7 @@ async def kick(
     async def kick_to_flow(
         self,
         message: BrokerMessage,
-        flow: Flow | None = None,
+        flow: FlowProtocol | None = None,
     ) -> None:
         """
         Send message to a flow-aware broker.
@@ -401,8 +401,10 @@ def inner(
 
     def register_task(
         self,
-        func: Callable[_FuncParams, _ReturnType]
-        | TaskDefinition[_FuncParams, _ReturnType],
+        func: (
+            Callable[_FuncParams, _ReturnType]
+            | TaskDefinition[_FuncParams, _ReturnType]
+        ),
         task_name: str | None = None,
         **labels: Any,
     ) -> AsyncTaskiqDecoratedTask[_FuncParams, _ReturnType]:

taskiq/brokers/inmemory_broker.py

@@ -9,7 +9,7 @@
 from taskiq.depends.progress_tracker import TaskProgress
 from taskiq.events import TaskiqEvents
 from taskiq.exceptions import UnknownTaskError
-from taskiq.flow import Flow
+from taskiq.flow import FlowProtocol
 from taskiq.message import BrokerMessage
 from taskiq.receiver import Receiver
 from taskiq.router import TaskiqRouter
@@ -135,7 +135,7 @@ def __init__(
         *,
         router: TaskiqRouter | None = None,
         broker_name: str | None = None,
-        default_flow: Flow | None = None,
+        default_flow: FlowProtocol | None = None,
     ) -> None:
         super().__init__(
             router=router,

taskiq/flow.py

@@ -1,58 +1,42 @@
-import enum
+from collections.abc import Mapping
 from dataclasses import dataclass, field, replace
-from typing import Any
+from typing import Protocol, runtime_checkable
 
-__all__ = ("Flow", "FlowKind")
+__all__ = ("Flow", "FlowProtocol")
 
 
-@enum.unique
-class FlowKind(str, enum.Enum):
-    """Transport-neutral flow shape."""
+@runtime_checkable
+class FlowProtocol(Protocol):
+    """Transport-neutral flow contract accepted by routers and brokers."""
 
-    QUEUE = "queue"
-    TOPIC = "topic"
-    STREAM = "stream"
+    name: str
+
+    def broker_options(self, broker_name: str) -> Mapping[str, object]:
+        """Return options relevant for a concrete broker implementation."""
+        ...
 
 
 @dataclass(frozen=True, slots=True)
 class Flow:
-    """Transport-neutral publish or subscribe address.
-
-    Plain flows are intentionally generic. Every broker may interpret a flow
-    using its own defaults: queue name, topic, stream, channel, list key, or any
-    other transport address.
+    """Generic transport-neutral flow address.
 
-    Broker packages can subclass this value object to expose transport-specific
-    details while still accepting plain Flow instances.
+    Broker packages can provide their own flow objects that implement
+    FlowProtocol, for example RabbitQueue, KafkaTopic, NatsSubject or RedisQueue.
+    This generic value object is intentionally small and works as a common
+    fallback for brokers that only need a named address with optional metadata.
     """
 
     name: str
-    kind: FlowKind = FlowKind.QUEUE
-    options: dict[str, Any] = field(
+    options: Mapping[str, object] = field(
         default_factory=dict,
         compare=False,
         hash=False,
     )
 
-    @classmethod
-    def queue(cls, name: str, **options: Any) -> "Flow":
-        """Create a queue-like flow."""
-        return cls(name=name, kind=FlowKind.QUEUE, options=options)
-
-    @classmethod
-    def topic(cls, name: str, **options: Any) -> "Flow":
-        """Create a topic-like flow."""
-        return cls(name=name, kind=FlowKind.TOPIC, options=options)
-
-    @classmethod
-    def stream(cls, name: str, **options: Any) -> "Flow":
-        """Create a stream-like flow."""
-        return cls(name=name, kind=FlowKind.STREAM, options=options)
-
-    def with_options(self, **options: Any) -> "Flow":
+    def with_options(self, **options: object) -> "Flow":
         """Return the same flow with additional generic options."""
         return replace(self, options={**self.options, **options})
 
-    def broker_options(self, broker_name: str) -> dict[str, Any]:
+    def broker_options(self, broker_name: str) -> Mapping[str, object]:
         """Return transport options for broker-specific implementations."""
         return dict(self.options)

taskiq/kicker.py

@@ -21,7 +21,7 @@
 from taskiq.abc.result_backend import AsyncResultBackend
 from taskiq.compat import model_dump
 from taskiq.exceptions import SendTaskError
-from taskiq.flow import Flow
+from taskiq.flow import FlowProtocol
 from taskiq.labels import prepare_label
 from taskiq.message import TaskiqMessage
 from taskiq.router import TaskiqRouter
@@ -74,7 +74,7 @@ def __init__(
         self.custom_schedule_id: str | None = None
         self.return_type = return_type
         self.route_broker: AsyncBroker | str | None = None
-        self.route_flow: Flow | None = None
+        self.route_flow: FlowProtocol | None = None
 
     def with_labels(
         self,
@@ -137,7 +137,7 @@ def with_broker(
 
     def with_flow(
         self,
-        flow: Flow | None,
+        flow: FlowProtocol | None,
     ) -> AsyncKicker[_FuncParams, _ReturnType]:
         """
         Replace flow for the current invocation.
@@ -151,12 +151,14 @@ def with_flow(
     def with_route(
         self,
         broker: AsyncBroker | str,
-        flow: Flow | None,
+        flow: FlowProtocol | None,
     ) -> AsyncKicker[_FuncParams, _ReturnType]:
         """
         Replace broker and flow for the current invocation.
 
         :param broker: broker instance or broker name.
+            Broker instances are preferred; names are kept for configuration
+            and backward-compatible lookup.
         :param flow: flow to send message to.
         :return: Kicker with a route override.
         """