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.

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
class MessageBusQueryFetcher[QueryPayloadType, QueryFetcherResult](
    QueryFetcherPort[QueryPayloadType, QueryFetcherResult]
):
    """Infrastructure adapter that fetches query results via a ``MessageBusPort``.

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

    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 
26
27
28
async def fetch(self, query: Query[QueryPayloadType]) -> QueryFetcherResult:
    """Fetch a query result via the message bus."""
    return await self._message_bus.dispatch(query)