Skip to content

Message Handler Port

message_handler_port

Inbound message-handling ports for the application blocks.

This module defines the contract for handling application messages
(commands, queries, and domain events). Message handlers sit at the
application boundary and orchestrate domain logic by invoking domain
services, aggregates, value objects, and outbound ports.

Responsibilities
  • Process an incoming message of a specific type.
  • Delegate work to domain logic and outbound ports.
  • Execute asynchronously.
Non-Responsibilities
  • Infrastructure concerns (transport, serialization, queues).
  • Transaction management (handled externally, e.g., by Unit of Work).
  • Persistence (handled by repositories).

MessageHandlerPort

Bases: InboundPort

Inbound port for handling messages asynchronously.

A MessageHandlerPort defines the contract for processing a single message
and producing a typed result. Implementations must avoid infrastructure
dependencies and remain purely application-blocks logic.

Implementers typically
  • Validate or transform the incoming message.
  • Orchestrate domain operations.
  • Invoke repositories, outbound ports, and publish domain events.
Example 
class CreateOrder(Command[dict[str, object]]):
    product_id: str


class CreateOrderHandler(MessageHandlerPort[CreateOrder, None]):
    async def handle(self, message: CreateOrder) -> None: ...
Source code in src/forging_blocks/application/ports/inbound/message_handler_port.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class MessageHandlerPort[MessageType, MessageHandlerResultType](
    InboundPort,
):
    """Inbound port for handling messages asynchronously.

    A MessageHandlerPort defines the contract for processing a single message
    and producing a typed result. Implementations must avoid infrastructure
    dependencies and remain purely application-blocks logic.

    Implementers typically:
        - Validate or transform the incoming message.
        - Orchestrate domain operations.
        - Invoke repositories, outbound ports, and publish domain events.

    Example:
        ```python
        class CreateOrder(Command[dict[str, object]]):
            product_id: str


        class CreateOrderHandler(MessageHandlerPort[CreateOrder, None]):
            async def handle(self, message: CreateOrder) -> None: ...
        ```
    """

    @abstractmethod
    async def handle(self, message: MessageType) -> MessageHandlerResultType:
        """Process a message and return an application result.

        Args:
            message: The message instance to handle.

        Returns:
            A value representing the result of the message processing.

        Raises:
            ApplicationError: If processing fails due to business logic violations.

        Notes:
            This method is asynchronous and should not block. Infrastructure
            concerns such as retry logic or message acknowledgment must be
            handled externally.

        """
        ...

handle(message: MessageType) -> MessageHandlerResultType abstractmethod async

Process a message and return an application result.

Parameters:

Name Type Description Default
message MessageType

The message instance to handle.

 required

Returns:

Type Description
MessageHandlerResultType

A value representing the result of the message processing.

Raises:

Type Description
ApplicationError

If processing fails due to business logic violations.
Notes

This method is asynchronous and should not block. Infrastructure
concerns such as retry logic or message acknowledgment must be
handled externally.

Source code in src/forging_blocks/application/ports/inbound/message_handler_port.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@abstractmethod
async def handle(self, message: MessageType) -> MessageHandlerResultType:
    """Process a message and return an application result.

    Args:
        message: The message instance to handle.

    Returns:
        A value representing the result of the message processing.

    Raises:
        ApplicationError: If processing fails due to business logic violations.

    Notes:
        This method is asynchronous and should not block. Infrastructure
        concerns such as retry logic or message acknowledgment must be
        handled externally.

    """
    ...

CommandHandlerPort

Bases: MessageHandlerPort[Command[CommandPayloadType], None]

Inbound port for handling commands.

Commands are fire-and-forget: the handler returns None on
success and raises on failure.

Example 
class CreateOrder(Command[dict[str, object]]):
    def __init__(self) -> None:
        super().__init__()

    @property
    def _payload(self) -> dict[str, object]:
        return {}


class CreateOrderHandler(CommandHandlerPort[dict[str, object]]):
    async def handle(self, message: Command[dict[str, object]]) -> None: ...
Source code in src/forging_blocks/application/ports/inbound/message_handler_port.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class CommandHandlerPort[CommandPayloadType](
    MessageHandlerPort[Command[CommandPayloadType], None],
):
    """Inbound port for handling commands.

    Commands are fire-and-forget: the handler returns ``None`` on
    success and raises on failure.

    Example:
        ```python
        class CreateOrder(Command[dict[str, object]]):
            def __init__(self) -> None:
                super().__init__()

            @property
            def _payload(self) -> dict[str, object]:
                return {}


        class CreateOrderHandler(CommandHandlerPort[dict[str, object]]):
            async def handle(self, message: Command[dict[str, object]]) -> None: ...
        ```

    """

QueryHandlerPort

Bases: MessageHandlerPort[Query[QueryPayloadType], QueryResultType]

Inbound port for handling queries.

Queries return a typed result; the handler must not mutate
observable state.

Example 
class GetOrder(Query[dict[str, object]]):
    def __init__(self) -> None:
        super().__init__()

    @property
    def _payload(self) -> dict[str, object]:
        return {}


class GetOrderHandler(QueryHandlerPort[dict[str, object], dict[str, object]]):
    async def handle(self, message: Query[dict[str, object]]) -> dict[str, object]: ...
Source code in src/forging_blocks/application/ports/inbound/message_handler_port.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class QueryHandlerPort[QueryPayloadType, QueryResultType](
    MessageHandlerPort[Query[QueryPayloadType], QueryResultType],
):
    """Inbound port for handling queries.

    Queries return a typed result; the handler must not mutate
    observable state.

    Example:
        ```python
        class GetOrder(Query[dict[str, object]]):
            def __init__(self) -> None:
                super().__init__()

            @property
            def _payload(self) -> dict[str, object]:
                return {}


        class GetOrderHandler(QueryHandlerPort[dict[str, object], dict[str, object]]):
            async def handle(self, message: Query[dict[str, object]]) -> dict[str, object]: ...
        ```

    """

EventHandlerPort

Bases: MessageHandlerPort[Event[EventPayloadType], None]

Inbound port for handling domain events.

Event handlers react to domain events; they return None and
may trigger side effects through outbound ports.

Example 
class OrderCreated(Event[dict[str, object]]):
    def __init__(self) -> None:
        super().__init__()

    @property
    def _payload(self) -> dict[str, object]:
        return {}


class OrderCreatedHandler(EventHandlerPort[dict[str, object]]):
    async def handle(self, message: Event[dict[str, object]]) -> None: ...
Source code in src/forging_blocks/application/ports/inbound/message_handler_port.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class EventHandlerPort[EventPayloadType](
    MessageHandlerPort[Event[EventPayloadType], None],
):
    """Inbound port for handling domain events.

    Event handlers react to domain events; they return ``None`` and
    may trigger side effects through outbound ports.

    Example:
        ```python
        class OrderCreated(Event[dict[str, object]]):
            def __init__(self) -> None:
                super().__init__()

            @property
            def _payload(self) -> dict[str, object]:
                return {}


        class OrderCreatedHandler(EventHandlerPort[dict[str, object]]):
            async def handle(self, message: Event[dict[str, object]]) -> None: ...
        ```

    """