Streaming

Streaming returns several typed chunks from one handler without first building a complete response in memory.

Compare the dispatch methods

StreamRequest is separate from Request. Type checkers therefore reject passing a stream request to send() or a single-response request to stream().

Define an asynchronous stream

Declare the chunk type on StreamRequest[ChunkT]. The handler type names the request, and the handler's return annotation names the iterator and chunk type:

import asyncio
from collections.abc import AsyncIterator
from dataclasses import dataclass

from pymediate import Mediator, Services, StreamRequest, StreamRequestHandler


@dataclass(frozen=True)
class OrderLine:
    order_id: int
    item: str
    quantity: int


@dataclass(frozen=True)
class ReadOrderLines(StreamRequest[OrderLine]):
    order_id: int


class ReadOrderLinesHandler(StreamRequestHandler[ReadOrderLines]):
    def __init__(self, rows: tuple[OrderLine, ...]) -> None:
        self._rows = rows

    async def __call__(self, request: ReadOrderLines) -> AsyncIterator[OrderLine]:
        for row in self._rows:
            if row.order_id == request.order_id:
                yield row


async def main() -> None:
    rows = (
        OrderLine(order_id=42, item="tea", quantity=2),
        OrderLine(order_id=42, item="coffee", quantity=1),
    )
    services = Services(ReadOrderLinesHandler(rows))
    mediator = Mediator(services)

    async for line in mediator.stream(ReadOrderLines(order_id=42)):
        print(line.item, line.quantity)


asyncio.run(main())

Do not await mediator.stream(...). The asynchronous method returns an AsyncIterator[ChunkT] directly; consume it with async for. The synchronous mediator returns Iterator[ChunkT] and uses for.

The OrderLine type flows from StreamRequest[OrderLine] to the iterator returned by stream(), so a type checker can infer the type of line at the call site.

The handler is resolved before iteration; chunks are produced lazily

stream() resolves the handler class and its service instance immediately. These errors are therefore raised at the stream() call:

• HandlerNotFoundError when no handler class is registered for the exact stream request type; and
• ServiceNotFoundError when that handler class has no instance in the service provider.

The handler's generator body runs only when the caller requests a chunk. Exceptions raised inside the generator therefore surface during iteration. Stopping iteration also stops requesting further chunks.

If the caller may break early and prompt generator cleanup matters, close the iterator explicitly. Use contextlib.aclosing() for the asynchronous API:

from contextlib import aclosing

async with aclosing(mediator.stream(ReadOrderLines(order_id=42))) as lines:
    async for line in lines:
        print(line.item, line.quantity)
        if line.item == "tea":
            break

The synchronous equivalent is with closing(mediator.stream(...)) as lines: from contextlib. Put resource acquisition and release in the handler or its underlying iterator so closing the generator releases those resources.

Handler validation

PyMediate validates a stream handler when Python defines the class:

• The request type must inherit from StreamRequest[ChunkT].
• The parameter annotation must be the exact request class named in StreamRequestHandler[...].
• An asynchronous handler must be an async generator and return AsyncIterator[ChunkT].
• A synchronous handler must be a generator and return Iterator[ChunkT].
• The iterator's chunk type must equal the type declared by the request.

A function that returns an existing iterator without using yield is not a stream handler under this contract. Delegate with async for and yield in the asynchronous version, or with yield from in the synchronous version.

PyMediate validates annotations without consuming the stream. It does not inspect each value yielded at runtime, so static checking remains responsible for detecting a generator that produces the wrong value type.

Scope and alternatives

Pipeline behaviors wrap send() only and do not run on stream(). Put logging, timing, or resource handling for a stream in its handler or in the source adapter.

Use Request[list[Item]] instead when the complete result is small and the caller needs it as one value. Use a broker or job system when chunks must survive process failure or be delivered to another process; PyMediate streams are in-process iterators.

Continue

• Notifications covers zero-or-more subscriber dispatch.
• Async and sync compares the iterator forms.
• StreamRequestHandler is the API reference.