Skip to content

Message Bus Query Fetcher

message_bus_query_fetcher

Message-bus-backed QueryFetcherPort adapter.

Delegates query dispatch to an injected MessageBusPort.

MessageBusQueryFetcher

Bases: QueryFetcherPort[QueryPayloadType, QueryFetcherResult]

Infrastructure adapter that fetches query results via a MessageBusPort.

Implements QueryFetcherPort by delegating fetch to
MessageBusPort.dispatch.

Example 
class Query[T]:
    def __init__(self) -> None:
        pass


class InMemoryMessageBus[MT, R]:
    async def dispatch(self, message: MT) -> R: ...
    def register(self, message_type, handler): ...


class MyQuery(Query[dict[str, object]]):
    def __init__(self) -> None:
        pass


bus = InMemoryMessageBus[MyQuery, str]()
fetcher = MessageBusQueryFetcher[dict[str, object], str](bus)
result = await fetcher.fetch(MyQuery())
Source code in src/forging_blocks/infrastructure/message_bus/message_bus_query_fetcher.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class MessageBusQueryFetcher[QueryPayloadType, QueryFetcherResult](
    QueryFetcherPort[QueryPayloadType, QueryFetcherResult]
):
    """Infrastructure adapter that fetches query results via a ``MessageBusPort``.

    Implements ``QueryFetcherPort`` by delegating ``fetch`` to
    ``MessageBusPort.dispatch``.

    Example:
        ```python
        class Query[T]:
            def __init__(self) -> None:
                pass


        class InMemoryMessageBus[MT, R]:
            async def dispatch(self, message: MT) -> R: ...
            def register(self, message_type, handler): ...


        class MyQuery(Query[dict[str, object]]):
            def __init__(self) -> None:
                pass


        bus = InMemoryMessageBus[MyQuery, str]()
        fetcher = MessageBusQueryFetcher[dict[str, object], str](bus)
        result = await fetcher.fetch(MyQuery())
        ```
    """

    def __init__(
        self,
        message_bus: MessageBusPort[Query[QueryPayloadType], QueryFetcherResult],
    ) -> None:
        self._message_bus = message_bus

    async def fetch(self, query: Query[QueryPayloadType]) -> QueryFetcherResult:
        """Fetch a query result via the message bus."""
        return await self._message_bus.dispatch(query)

fetch(query: Query[QueryPayloadType]) -> QueryFetcherResult async

Fetch a query result via the message bus.

Source code in src/forging_blocks/infrastructure/message_bus/message_bus_query_fetcher.py 
48
49
50
async def fetch(self, query: Query[QueryPayloadType]) -> QueryFetcherResult:
    """Fetch a query result via the message bus."""
    return await self._message_bus.dispatch(query)