Command Sender Port

command_sender_port

Outbound port for asynchronously sending commands.

Defines the CommandSenderPort protocol 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[Command[CommandPayloadType], None], Protocol

Protocol for dispatching command messages asynchronously.

Any object with an async send method that accepts a Command
satisfies this protocol — no inheritance required.

Source code in src/forging_blocks/application/ports/outbound/command_sender_port.py

class CommandSenderPort[CommandPayloadType](
    OutboundPort[Command[CommandPayloadType], None],
    Protocol,
):
    """Protocol for dispatching command messages asynchronously.

    Any object with an async ``send`` method that accepts a ``Command``
    satisfies this protocol — no inheritance required.
    """

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

        Args:
            command: The command instance to dispatch.
        """
        ...

send(command: Command[CommandPayloadType]) -> None 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

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

    Args:
        command: The command instance to dispatch.
    """
    ...