Event Publisher Port

event_publisher_port

Outbound port for asynchronously publishing domain events.

Defines the EventPublisherPort protocol for publishing domain events
to an external message bus or event broker. Infrastructure implementations
determine durability, ordering, and delivery guarantees.

Responsibilities

• Publish domain events after domain changes occur.

Non-Responsibilities

• Persist events.
• Guarantee ordering or durability.

EventPublisherPort

Bases: OutboundPort[Event[EventPayloadType], None], Protocol

Protocol for publishing domain events asynchronously.

Any object with an async publish method that accepts an Event
satisfies this protocol — no inheritance required.

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

class EventPublisherPort[EventPayloadType](
    OutboundPort[Event[EventPayloadType], None],
    Protocol,
):
    """Protocol for publishing domain events asynchronously.

    Any object with an async ``publish`` method that accepts an ``Event``
    satisfies this protocol — no inheritance required.
    """

    async def publish(self, event: Event[EventPayloadType]) -> None:
        """Publish a domain event.

        Args:
            event: The domain event to publish.
        """
        ...

publish(event: Event[EventPayloadType]) -> None async

Publish a domain event.

Parameters:

Name	Type	Description	Default
event	Event[EventPayloadType]	The domain event to publish.	required

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

async def publish(self, event: Event[EventPayloadType]) -> None:
    """Publish a domain event.

    Args:
        event: The domain event to publish.
    """
    ...