Skip to content

Command Sender Port

command_sender_port

Outbound port for asynchronously sending commands.

Defines the CommandSenderPort ABC for dispatching commands via
an external message bus. Infrastructure implementations determine
delivery semantics.

Responsibilities
  • Abstract command dispatch from transport details.
Non-Responsibilities
  • Delivery guarantees (at-least-once, ordering, retries).
  • Serialization or transport concerns.

CommandSenderPort

Bases: OutboundPort

ABC for dispatching command messages asynchronously.

Responsibilities
  • Accept any domain command type and dispatch it asynchronously.
  • Maintain a contract decoupled from any specific transport or
    message bus implementation.
Non-Responsibilities
  • Validate command payloads — that belongs to the validation layer.
  • Implement retry, delivery guarantees, or dead-letter handling.
  • Handle registration — registrations happen through a message bus,
    not through this port.
Example 
sender = MyCommandSender[PlaceOrderPayload]()
await sender.send(PlaceOrder(order_id="42"))
Source code in src/forging_blocks/application/ports/outbound/command_sender_port.py 
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
51
class CommandSenderPort[CommandPayloadType](
    OutboundPort,
):
    """ABC for dispatching command messages asynchronously.

    Responsibilities:
        - Accept any domain command type and dispatch it asynchronously.
        - Maintain a contract decoupled from any specific transport or
          message bus implementation.

    Non-Responsibilities:
        - Validate command payloads — that belongs to the validation layer.
        - Implement retry, delivery guarantees, or dead-letter handling.
        - Handle registration — registrations happen through a message bus,
          not through this port.

    Example:
        ```python
        sender = MyCommandSender[PlaceOrderPayload]()
        await sender.send(PlaceOrder(order_id="42"))
        ```
    """

    @abstractmethod
    async def send(self, command: Command[CommandPayloadType]) -> None:
        """Send a command asynchronously.

        Args:
            command: The command instance to dispatch.

        """

send(command: Command[CommandPayloadType]) -> None abstractmethod async

Send a command asynchronously.

Parameters:

Name Type Description Default
command Command[CommandPayloadType]

The command instance to dispatch.

 required
Source code in src/forging_blocks/application/ports/outbound/command_sender_port.py 
44
45
46
47
48
49
50
51
@abstractmethod
async def send(self, command: Command[CommandPayloadType]) -> None:
    """Send a command asynchronously.

    Args:
        command: The command instance to dispatch.

    """