Event-Driven Patterns

Domain events, the EventPublisher port, adapter selection, and lifecycle rules.

Show:

Framework implementation: For FastAPI see 18-framework-implementations/fastapi/10-event-publishing.md.

Why Domain Events

Domain events represent significant occurrences within a bounded context — state changes that other parts of the system (or other bounded contexts) may need to react to. They decouple the producer from any downstream reactions: the account service publishes account.created; it does not know or care what happens next.

Events are facts — immutable records of what happened. They are named in past tense and carry all the data consumers need to react without querying back.

Examples:

account.created — triggers welcome email, downstream provisioning
message.sent — triggers notification, analytics pipeline
order.confirmed — triggers fulfilment, billing

The EventPublisher Port

Define an abstract interface in the application layer. All concrete adapters implement it:

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import uuid
 
 
@dataclass
class DomainEvent:
    event_type: str                         # dot-namespaced, past tense: "account.created"
    payload: dict[str, Any]                 # all data consumers need — no follow-up fetches
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
 
 
class EventPublisher(ABC):
    @abstractmethod
    async def publish(self, event: DomainEvent) -> None:
        """Publish a domain event to the configured broker."""
 
    async def startup(self) -> None:
        """Called at application startup. Override to initialise connections."""
 
    async def shutdown(self) -> None:
        """Called at shutdown. Override to flush buffered events."""

The interface lives in application/events/publisher.py. Infrastructure adapters live in infrastructure/events/adapters/. Application and domain code never reference a concrete adapter.

Adapters

Adapter	When to use
In-memory	Tests and local development — collects events in a list, zero external deps
SNS	Production on AWS — fan-out to multiple consumers via topic subscriptions
Celery	Production — task-queue-based delivery, works with any broker
No-op	Services that don't publish events yet — satisfies the interface safely

The adapter is selected by configuration (e.g. settings.event_publisher_type) and wired at startup via a factory function. The factory reads the setting and returns the appropriate concrete instance.

DomainEvent Structure

Field	Type	Description
`event_type`	`str`	Dot-namespaced, past tense: `account.created`, `message.sent`
`payload`	`dict`	All data consumers need — no IDs requiring a follow-up fetch
`event_id`	`str`	UUID; idempotency key for consumers
`timestamp`	`str`	ISO 8601 UTC; when the event occurred (not when published)

Events must be self-contained: consumers should not need to query the originating service to understand and act on the event.

Publishing in Application Handlers

Application handlers publish events after a successful operation — not inside domain entities:

async def handle(self, command: CreateAccountCommand) -> Account:
    account = Account.create(phone=command.phone)
    await self.uow.accounts.add(account)
    await self.uow.commit()                         # persist first
 
    await self.event_publisher.publish(DomainEvent(
        event_type="account.created",
        payload={"account_id": str(account.id), "phone": account.phone},
    ))
 
    return account

Domain entities never depend on EventPublisher — they enforce invariants and may expose domain events as a list attribute, but delivery is the application handler's responsibility.

Lifecycle

The event publisher is initialised at application startup (startup()) and shut down gracefully on exit (shutdown()). Adapters that buffer (e.g. batching outbound calls) must flush on shutdown to avoid losing events.

Rules

EventPublisher and DomainEvent are defined in the application layer — never in infrastructure.
Concrete adapters are in infrastructure/events/adapters/.
Application handlers call publisher.publish() — domain entities do not.
Event type names use dot notation and past tense: account.created, not CreateAccount or AccountCreatedEvent.
Events carry all necessary data — consumers must not need to call back to the source service.
The in-memory adapter is always used in tests — never mock the publisher interface.
Publish after committing the unit of work — do not publish events for transactions that may roll back.

Framework implementations

Event publishing uses the ports & adapters pattern. The application layer defines an abstract EventPublisher interface. Infrastructure provides in-memory, SNS, and Celery adapters. The factory selects the adapter via settings.

Interface (`application/events/publisher.py`)

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from uuid import UUID
import datetime
 
 
@dataclass(frozen=True)
class DomainEvent:
    event_type: str
    aggregate_id: str
    aggregate_type: str
    payload: dict = field(default_factory=dict)
    request_id: str = ""
    occurred_at: datetime.datetime = field(
        default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
    )
 
 
class EventPublisher(ABC):
    @abstractmethod
    async def publish(self, event: DomainEvent) -> None: ...
 
    @abstractmethod
    async def publish_batch(self, events: list[DomainEvent]) -> None: ...
 
    async def start(self) -> None:
        """Optional: called on app startup."""
        pass
 
    async def shutdown(self) -> None:
        """Optional: called on app shutdown."""
        pass

Adapters

In-Memory (`infrastructure/events/adapters/inmemory.py`)

from application.events.publisher import EventPublisher, DomainEvent
 
 
class InMemoryEventPublisher(EventPublisher):
    def __init__(self):
        self.published: list[DomainEvent] = []
 
    async def publish(self, event: DomainEvent) -> None:
        self.published.append(event)
 
    async def publish_batch(self, events: list[DomainEvent]) -> None:
        self.published.extend(events)

Used in local development and all tests. The published list is inspectable in tests.

SNS (`infrastructure/events/adapters/sns.py`)

import json
import asyncio
import boto3
from application.events.publisher import EventPublisher, DomainEvent
 
 
class SNSEventPublisher(EventPublisher):
    def __init__(self, topic_arn: str, region: str, timeout: float = 5.0):
        self._topic_arn = topic_arn
        self._timeout = timeout
        self._client = boto3.client("sns", region_name=region)
 
    async def publish(self, event: DomainEvent) -> None:
        message = json.dumps({
            "event_type": event.event_type,
            "aggregate_id": event.aggregate_id,
            "aggregate_type": event.aggregate_type,
            "payload": event.payload,
            "request_id": event.request_id,
            "occurred_at": event.occurred_at.isoformat(),
        })
        await asyncio.wait_for(
            asyncio.to_thread(
                self._client.publish,
                TopicArn=self._topic_arn,
                Message=message,
                MessageAttributes={
                    "event_type": {
                        "DataType": "String",
                        "StringValue": event.event_type,
                    }
                },
            ),
            timeout=self._timeout,
        )
 
    async def publish_batch(self, events: list[DomainEvent]) -> None:
        for event in events:
            await self.publish(event)

Celery (`infrastructure/events/adapters/celery.py`)

from celery import Celery
from application.events.publisher import EventPublisher, DomainEvent
 
 
class CeleryEventPublisher(EventPublisher):
    def __init__(self, broker_url: str, task_name: str = "process_event"):
        self._app = Celery(broker=broker_url)
        self._task_name = task_name
 
    async def publish(self, event: DomainEvent) -> None:
        self._app.send_task(self._task_name, args=[{
            "event_type": event.event_type,
            "aggregate_id": event.aggregate_id,
            "payload": event.payload,
        }])
 
    async def publish_batch(self, events: list[DomainEvent]) -> None:
        for event in events:
            await self.publish(event)

Factory (`infrastructure/events/factories.py`)

from config.settings import Settings
from application.events.publisher import EventPublisher
from .adapters.inmemory import InMemoryEventPublisher
from .adapters.sns import SNSEventPublisher
from .adapters.celery import CeleryEventPublisher
 
 
def build_event_publisher(settings: Settings) -> EventPublisher:
    match settings.event_adapter:
        case "inmemory":
            return InMemoryEventPublisher()
        case "sns":
            return SNSEventPublisher(
                topic_arn=settings.event_sns_topic_arn,
                region=settings.aws_region,
                timeout=settings.event_publish_timeout,
            )
        case "celery":
            return CeleryEventPublisher(
                broker_url=settings.event_celery_broker_url,
                task_name=settings.event_celery_task_name,
            )
        case _:
            raise ValueError(f"Unknown event adapter: {settings.event_adapter}")

Usage in Application Handlers

# application/account/create.py
from application.events.publisher import EventPublisher, DomainEvent
 
 
class CreateAccountHandler:
    def __init__(self, repo, uow, event_publisher: EventPublisher):
        self._repo = repo
        self._uow = uow
        self._publisher = event_publisher
 
    async def execute(self, command, request_id: str = "") -> Account:
        async with self._uow as session:
            account = Account.create(...)
            await self._repo.save(session, account)
 
        # Publish after commit — not inside the UoW context
        await self._publisher.publish(DomainEvent(
            event_type="account.created",
            aggregate_id=str(account.id),
            aggregate_type="Account",
            payload={"email": account.email},
            request_id=request_id,
        ))
        return account

Publish after commit — events are published after the UoW commits to avoid emitting events for transactions that roll back.

Lifecycle in `api/lifespan.py`

from contextlib import asynccontextmanager
from api.dependencies import get_event_publisher
 
@asynccontextmanager
async def lifespan(app):
    publisher = get_event_publisher()
    await publisher.start()
    yield
    await publisher.shutdown()

start() and shutdown() are no-ops for in-memory and SNS adapters. For Celery or any adapter with background tasks, they handle connection and cleanup.

Rules

Application handlers publish events after the UoW commits, never inside the transaction.
The in-memory adapter's published list is the test assertion surface — check publisher.published in tests.
Event event_type is dot-separated past tense: account.created, message.sent, enrollment.cancelled.
request_id is always passed through to events for end-to-end traceability.

Unit of Work Pattern Actor Context

Event-Driven Patterns

Why Domain Events

The EventPublisher Port

Adapters

DomainEvent Structure

Publishing in Application Handlers

Lifecycle

Rules

Interface (application/events/publisher.py)

Adapters

In-Memory (infrastructure/events/adapters/inmemory.py)

SNS (infrastructure/events/adapters/sns.py)

Celery (infrastructure/events/adapters/celery.py)

Factory (infrastructure/events/factories.py)

Usage in Application Handlers

Lifecycle in api/lifespan.py

Rules

Interface (`application/events/publisher.py`)

In-Memory (`infrastructure/events/adapters/inmemory.py`)

SNS (`infrastructure/events/adapters/sns.py`)

Celery (`infrastructure/events/adapters/celery.py`)

Factory (`infrastructure/events/factories.py`)

Lifecycle in `api/lifespan.py`