Skip to content

Event Publisher Port

event_publisher_port

Outbound port for asynchronously publishing domain events.

Defines the EventPublisherPort contract 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

ABC for publishing domain events asynchronously.

Example 
publisher = MyEventPublisher[OrderEventData]()
await publisher.publish(OrderPlaced(order_id="42"))
Source code in src/forging_blocks/application/ports/outbound/event_publisher_port.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class EventPublisherPort[EventPayloadType](
    OutboundPort,
):
    """ABC for publishing domain events asynchronously.

    Example:
        ```python
        publisher = MyEventPublisher[OrderEventData]()
        await publisher.publish(OrderPlaced(order_id="42"))
        ```
    """

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

        Args:
            event: The domain event to publish.

        """

publish(event: Event[EventPayloadType]) -> None abstractmethod 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 
33
34
35
36
37
38
39
40
@abstractmethod
async def publish(self, event: Event[EventPayloadType]) -> None:
    """Publish a domain event.

    Args:
        event: The domain event to publish.

    """