Streaming Fallback

• Back to Testing Overview

  Return to the Testing overview page with all topics.

  Back to Overview

yields and invocations show both handlers. Values that were already yielded stay.

See Stream Handling Fallback.

def streaming_mapper(mapper: cqrs.RequestMap) -> None:
    mapper.bind(
        ExportOrders,
        cqrs.RequestHandlerFallback(ExportOrdersHandler, ExportOrdersFallback),
    )

items = [item async for item in harness.streaming_mediator.stream(ExportOrders(batch=2))]
harness.invocations.assert_entered(ExportOrdersHandler)
harness.invocations.assert_entered(ExportOrdersFallback)
assert [item.index for item in harness.yields] == [0, 0, 1]

Library example: examples/testing/streaming.py.