Event Handlers

DDD CQRS ES

When an aggregate changes state, other parts of the system often need to react, updating a different aggregate, sending a notification, or triggering a downstream process. Putting that logic inside the originating aggregate would violate its boundary. Event handlers solve this: they listen for domain events and execute side effects in their own transaction, keeping aggregates decoupled.

Event handlers consume events raised in an aggregate and help sync the state of the aggregate with other aggregates and other systems. An event handler stays with the aggregate that raised the event. When another aggregate must change, the handler issues a command, and that aggregate's command handler does the write.

Choosing the Right Consumer

Protean provides four elements that consume events or messages. Each serves a distinct purpose:

Element	Consumes	Source	Association	Dispatch	Use case
Event Handler	Domain events	Internal event store	part_of (aggregate)	@handle(EventClass)	Side effects, cross-aggregate sync
Projector	Domain events	Internal event store	projector_for (projection)	@on(EventClass)	Maintain read models (projections)
Subscriber	Raw dict payloads	External message broker	stream (broker stream)	__call__(payload)	Anti-corruption layer for external systems
Query Handler	Queries	Synchronous dispatch	part_of (projection)	@read(QueryClass)	Structured read-side access

Use event handlers when you need to react to internal domain events with side effects (syncing aggregates, sending notifications, triggering processes). Use projectors when the reaction is specifically maintaining a read model. Use subscribers when messages come from outside your bounded context. Use query handlers when you need structured, validated read access to projections.

Defining an Event Handler

Event Handlers are defined with the Domain.event_handler decorator. Below is a simplified example of an Event Handler that syncs stock levels in Inventory in response to changes in the Order aggregate.

from protean import Domain, current_domain, handle
from protean.fields import Identifier, Integer, List, String

domain = Domain()
domain.config["event_processing"] = "sync"
domain.config["command_processing"] = "sync"


@domain.event(part_of="Order")
class OrderShipped:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)
    total_amount: Integer(required=True)


@domain.command(part_of="Inventory")
class ReduceStock:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.aggregate
class Order:
    book_id: Identifier(required=True)
    quantity: Integer(required=True)
    total_amount: Integer(required=True)
    status: String(choices=["PENDING", "SHIPPED", "DELIVERED"], default="PENDING")

    def ship_order(self):
        self.status = "SHIPPED"

        self.raise_(  # (1)
            OrderShipped(
                order_id=self.id,
                book_id=self.book_id,
                quantity=self.quantity,
                total_amount=self.total_amount,
            )
        )


@domain.aggregate
class Inventory:
    book_id: Identifier(required=True)
    in_stock: Integer(required=True)
    applied_order_ids: List(content_type=String)

    def reduce_stock(self, order_id: str, quantity: int) -> None:
        self.in_stock -= quantity
        self.applied_order_ids = [*self.applied_order_ids, order_id]


@domain.event_handler(part_of=Order)  # (2)
class ManageInventory:
    @handle(OrderShipped)
    def reduce_stock_level(self, event: OrderShipped):
        current_domain.process(
            ReduceStock(
                order_id=event.order_id,
                book_id=event.book_id,
                quantity=event.quantity,
            )
        )


@domain.command_handler(part_of=Inventory)
class InventoryCommandHandler:
    @handle(ReduceStock)
    def reduce_stock(self, command: ReduceStock):
        repo = current_domain.repository_for(Inventory)
        inventory = repo.find_by(book_id=command.book_id)

        if command.order_id in inventory.applied_order_ids:  # (3)
            return

        inventory.reduce_stock(command.order_id, command.quantity)  # (4)
        repo.add(inventory)


domain.init(traverse=False)
with domain.domain_context():
    # Persist Order
    order = Order(book_id="book-1", quantity=10, total_amount=100)
    domain.repository_for(Order).add(order)

    # Persist Inventory
    inventory = Inventory(book_id="book-1", in_stock=100)
    domain.repository_for(Inventory).add(inventory)

    # Ship Order
    order.ship_order()
    domain.repository_for(Order).add(order)

    # Verify that Inventory Level has been reduced
    stock = domain.repository_for(Inventory).get(inventory.id)
    print(stock.to_dict())
    assert stock.in_stock == 90

    # A repeated delivery of the same order leaves the stock alone
    domain.process(ReduceStock(order_id=order.id, book_id="book-1", quantity=10))
    assert domain.repository_for(Inventory).get(inventory.id).in_stock == 90

1. Order aggregate fires OrderShipped event on book being shipped.

2. Event handler is registered with part_of=Order, the aggregate that owns OrderShipped. It does not write to Inventory itself. It issues a ReduceStock command, which is part_of=Inventory, and Inventory's command handler does the write. Inventory changes only through its own command handler. This example sets command_processing to "sync", so the command runs inside the event handler's unit of work and commits with it. With async command processing, the command runs later in its own transaction.

3. The command carries the order id taken from the event. Events are delivered at least once, so the same OrderShipped can arrive twice. Inventory records the order ids it has applied. The command handler returns without changes when it has already seen this order id.

4. The command handler calls the reduce_stock() domain method, so Inventory changes its own state.

When one event starts several steps that depend on each other, use a process manager to coordinate them.

Simulating the example, we can see that the stock levels were decreased in response to the OrderShipped event.

In [1]: order = Order(book_id="book-1", quantity=10, total_amount=100)

In [2]: domain.repository_for(Order).add(order)
Out[2]: <Order: Order object (id: 635014d2-0471-49d9-bcdc-cc7f180d166f)>

In [3]: inventory = Inventory(book_id="book-1", in_stock=100)

In [4]: domain.repository_for(Inventory).add(inventory)
Out[4]: <Inventory: Inventory object (id: 79380336-c485-463a-af8d-6508e28eed13)>

In [5]: order.ship_order()

In [6]: domain.repository_for(Order).add(order)
Out[6]: <Order: Order object (id: 635014d2-0471-49d9-bcdc-cc7f180d166f)>

In [7]: stock = domain.repository_for(Inventory).get(inventory.id)

In [8]: stock.to_dict()
Out[8]: {
 'book_id': 'book-1',
 'in_stock': 90,
 'applied_order_ids': ['635014d2-0471-49d9-bcdc-cc7f180d166f'],
 'id': '79380336-c485-463a-af8d-6508e28eed13',
 '_version': 1
 }

Internal workflow

Event handlers follow an asynchronous, fire-and-forget pattern. When an event is published, event handlers process it without returning any values to the caller.

1. Aggregate Publishes Event: An action in an aggregate triggers an event to be published.
2. Domain Stores Event: The domain stores the event in the event store.
3. Event Store Confirms Storage: The event store confirms the event has been stored.
4. Domain Returns to Aggregate: The domain returns control to the aggregate.
5. Event Store Delivers Event: Asynchronously, the event store delivers the event to all subscribed event handlers.
6. Event Handler Processes Event: The event handler receives and processes the event.
7. Event Handler Loads/Updates Aggregates: If needed, the event handler loads and updates relevant aggregates.
8. Repository Returns Data: The repository returns requested data to the event handler.
9. Event Handler Performs Side Effects: The event handler may perform additional side effects (sending emails, updating other systems, etc.).
10. Event Handler Persists Data and Optionally Raises Events: The event handler persists the mutated aggregate, which can also raise events.

The @handle Decorator

The @handle decorator binds a handler method to a specific event type. Import it from protean:

from protean import handle

Each @handle-decorated method receives exactly one event class as its argument. When that event type arrives, the framework invokes the corresponding method with the deserialized event object.

A single event handler class can contain multiple @handle methods, each processing a different event type:

from protean import Domain, current_domain
from protean.fields import Identifier, Integer, String

domain = Domain(name="Bookshop")
domain.config["event_processing"] = "sync"
domain.config["command_processing"] = "sync"


@domain.event(part_of="Order")
class OrderPlaced:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.event(part_of="Order")
class OrderShipped:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.event(part_of="Order")
class OrderCancelled:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.aggregate
class Order:
    book_id: Identifier(required=True)
    quantity: Integer(required=True)
    status: String(default="PLACED")

    def place(self):
        self.raise_(
            OrderPlaced(order_id=self.id, book_id=self.book_id, quantity=self.quantity)
        )

    def ship(self):
        self.status = "SHIPPED"
        self.raise_(
            OrderShipped(order_id=self.id, book_id=self.book_id, quantity=self.quantity)
        )

    def cancel(self):
        self.status = "CANCELLED"
        self.raise_(
            OrderCancelled(
                order_id=self.id, book_id=self.book_id, quantity=self.quantity
            )
        )


@domain.command(part_of="Inventory")
class ReduceStock:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.command(part_of="Inventory")
class RestoreStock:
    order_id: Identifier(required=True)
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.aggregate
class Inventory:
    book_id: Identifier(required=True)
    in_stock: Integer(required=True)


@domain.event_handler(part_of=Order)
class ManageInventory:
    @handle(OrderShipped)
    def reduce_stock(self, event: OrderShipped):
        current_domain.process(
            ReduceStock(
                order_id=event.order_id,
                book_id=event.book_id,
                quantity=event.quantity,
            )
        )

    @handle(OrderCancelled)
    def restore_stock(self, event: OrderCancelled):
        current_domain.process(
            RestoreStock(
                order_id=event.order_id,
                book_id=event.book_id,
                quantity=event.quantity,
            )
        )

Each @handle method runs within its own Unit of Work. If the handler modifies an aggregate and persists it, the changes are committed atomically. If an error occurs, the transaction is rolled back.

Several methods can also handle the same event. They are independent reactions to one fact: each gets its own Unit of Work, and one that fails does not stop the others from running. Their order is unspecified, so nothing may depend on one running before another. When two steps must happen in order, put them in one method, or have the first raise an event that the second handles.

If more than one method fails while handling the same event, the failures are raised together as an ExceptionGroup. A single failure propagates as itself.

The same holds across handler classes. Two handler classes subscribing to one event are independent reactions too. Under asynchronous processing, each class has its own subscription. Under synchronous (event_processing="sync") processing, one class failing no longer stops the other.

A separate handler class also gets its own subscription and its own checkpoint, which is what makes it fully independent when it touches a different aggregate. When two classes both mutate the same aggregate, that independence causes contention: the two classes run concurrently and conflict on the aggregate's version, which triggers version-conflict retries. Keep those reactions as sibling methods on one class, where they run in sequence and each sees the previous commit. When the reactions depend on each other's writes, put them in one method instead. When those writes change another aggregate, that method is a command handler method on that aggregate. Calling External Systems from Handlers shows that recipe for a multi-facet sync. It also shows how to split a method that both persists data and calls an external system.

One event class per method

Each @handle method accepts exactly one event class. To handle multiple event types, define multiple methods in the same handler class. The one exception is the $any wildcard below, which matches every event.

Handle every event with $any

To process every event on the handler's stream with a single method, pass the wildcard string "$any" to @handle instead of a specific event class. This is useful for cross-cutting concerns such as an audit trail:

@domain.aggregate
class AuditEntry:
    event_type: String(required=True)
    payload: Dict()

    @classmethod
    def from_event(cls, event):
        return cls(event_type=event.__class__.__name__, payload=event.payload)


@domain.event_handler(part_of=Order)
class OrderAudit:
    @handle("$any")
    def record(self, event):
        current_domain.repository_for(AuditEntry).add(AuditEntry.from_event(event))

A $any handler stays scoped to its aggregate's stream: it receives all events for that aggregate, and only those. It fires under both synchronous (event_processing="sync") and asynchronous processing.

Return Values from Event Handlers

Event handlers in Protean follow the standard CQRS pattern where event handlers do not return values to the caller. Protean follows this pattern for three reasons:

1. Decoupling: The publisher of events remains completely decoupled from the consumers.
2. Asynchronous Processing: Events can be processed in the background without blocking.
3. Multiple Consumers: Multiple event handlers can process the same event independently.

If an event handler needs to communicate information as part of its processing, it should:

• Emit new events
• Update relevant aggregates that can be queried later
• Log information for monitoring purposes

Idempotency and at-least-once delivery

Event delivery is at-least-once: a handler can receive the same event more than once (after a crash-and-republish, or a redelivery when a handler committed but was not yet acknowledged). Write event handlers to be idempotent so a duplicate delivery is harmless: guard on state, upsert rather than accumulate, and treat "already applied" as a no-op. See Idempotent Event Handlers.

Unlike projectors and subscribers, event handlers have no built-in idempotent=True option: a handler may write to arbitrary aggregates across providers, so no single atomic dedup marker applies. Protean offers framework-managed deduplication only where it knows the one provider that holds the writes: a projector's projection provider (see Projectors) and an idempotent subscriber's provider (see Subscribers).

Configuration Options

Handler Options

• part_of: The aggregate to which the event handler is connected.

• stream_category: The event handler listens to events on this stream category. The stream category defaults to the category of the aggregate associated with the handler.

  When the handler reacts to its own aggregate's events, keep the default. To change another aggregate, do not point stream_category at the other aggregate's stream: check reports that handler as EVENT_HANDLER_FOREIGN_EVENT. Put the handler in the aggregate that owns the event and issue a command, as in the example above. Learn more in the Stream Categories guide.

• source_stream: When specified, the event handler only consumes events whose origin_stream matches this value. This option filters events by what originally triggered them. Use it when the same event type can be raised from different contexts and you only want to react to a specific trigger.

  @domain.event_handler(part_of=Order, source_stream="manage_order")
  class EmailNotifications:
      @handle(OrderShipped)
      def send_shipping_email(self, event: OrderShipped):
          # Only invoked when OrderShipped was triggered by a command
          # in the manage_order stream, not by a bulk import or replay.
          ...
  

• retries, backoff, retry_exceptions: Opt-in auto-retry on transient infrastructure exceptions, applied inside the handler before any subscription-level retry. See Transient-failure retries.

Set part_of

An event handler needs part_of or stream_category, or registration fails. Set part_of to the aggregate that raises the events the handler reacts to. This association determines the default stream category.

Subscription Options

You can set subscription options on an event handler. These options control how the Protean server consumes messages:

• subscription_type: Type of subscription to use:

  • "stream": Uses Redis Streams with consumer groups (recommended for production)
  • "event_store": Reads directly from event store (for projections/replay)

• subscription_profile: Pre-configured profile for common scenarios:

  • "production": High throughput with reliability guarantees
  • "fast": Low-latency processing
  • "batch": High-volume batch processing
  • "debug": Development and debugging
  • "projection": Building read models (uses event_store type)
  • The name of a custom profile defined under [server.profiles]

• subscription_config: Dictionary of specific configuration options:

  • messages_per_tick: Messages to process per batch
  • blocking_timeout_ms: Timeout for blocking reads (stream only)
  • max_retries: Retry attempts before DLQ (stream only)
  • enable_dlq: Enable dead letter queue (stream only)
  • position_update_interval: Position update frequency (event_store only)

Example with Subscription Configuration

@domain.event_handler(
    part_of=Order,
    subscription_profile="production",
    subscription_config={
        "messages_per_tick": 100,
        "enable_dlq": True,
    },
)
class OrderEventHandler:
    @handle(OrderPlaced)
    def send_confirmation(self, event): ...

Retry and Dead Letter Queue (DLQ)

When using Redis Streams (subscription_type="stream"), Protean supports automatic retries and dead-letter queuing for failed messages:

• max_retries (default: 3): The number of attempts in total, counting the first, before a failed message is moved to the dead letter queue. With the default, a message is tried 3 times. Each retry re-delivers the message from the pending entries list (PEL).

• enable_dlq (default: True): Messages that exhaust their retry budget are moved to a separate dead-letter stream (<stream>:dlq). You can inspect and reprocess DLQ messages manually or with a scheduled job. Set enable_dlq to False to drop those messages instead.

@domain.event_handler(
    part_of=Order,
    subscription_type="stream",
    subscription_config={
        "max_retries": 5,
        "enable_dlq": True,
    },
)
class CriticalOrderHandler:
    @handle(OrderPlaced)
    def process_order(self, event: OrderPlaced):
        # If this fails 5 times, the message moves to the DLQ
        ...

Event store subscriptions

The settings above apply to subscription_type="stream" (Redis Streams). An event store subscription keeps no dead-letter stream. It records each failed position and retries it on a later recovery pass, up to max_retries (default 3), set under [server.event_store_subscription]. A position that still fails is marked exhausted and skipped. The exhausted positions are the event store's dead-letter queue. You list, inspect, replay, and purge them with protean eventstore dlq.

See Server → Configuration for detailed configuration options and the priority hierarchy.

Transient-failure retries

Independently of subscription-level retry/DLQ, a handler can retry in place when it fails with a transient infrastructure exception (a dropped connection, a timeout). These fast retries happen inside the handler, before the message is ever re-delivered or routed to the DLQ:

@domain.event_handler(part_of=Order, retries=3, backoff="exponential")
class InventorySync:
    @handle(OrderPlaced)
    def reserve_stock(self, event: OrderPlaced):
        # A ConnectionError here is retried up to 3 times with exponential
        # backoff before the failure surfaces to the subscription.
        ...

Each attempt runs in a fresh Unit of Work. Only genuinely transient exceptions are retried; by default ConnectionError, TimeoutError, and SendError, overridable with retry_exceptions. The behaviour is identical to command handlers. The retry is opt-in and disabled unless retries is set or server.transient_retry is enabled. This is distinct from the version-conflict (OCC) retry that handles ExpectedVersionError. See server.transient_retry for domain-wide defaults.

Error Handling

Protean provides error handling for event handlers through the optional handle_error method. This method allows event handlers to gracefully recover from exceptions without disrupting the overall event processing pipeline.

The handle_error Method

You can add a handle_error class method to your event handler to implement custom error handling:

import logging

logger = logging.getLogger(__name__)


@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderShipped)
    def reduce_stock_for_order(self, event):
        # Event handling logic that might raise exceptions
        ...

    @classmethod
    def handle_error(cls, exc: Exception, message):
        """Custom error handling for event processing failures"""
        # Log the error
        logger.error(f"Failed to process event {message.metadata.headers.type}: {exc}")

        # Perform recovery operations
        # Example: store failed events for retry, trigger compensating actions, etc.

How It Works

1. When an exception occurs during event processing, the Protean Engine catches it.
2. The engine logs detailed error information including stack traces.
3. The engine calls the event handler's handle_error method, passing:
4. The original exception that was raised
5. The event message being processed when the exception occurred
6. After handle_error completes, processing continues with subsequent events.

Error Handler Failures

If an exception occurs within the handle_error method itself, the Protean Engine catches and logs that exception as well, so the event processing pipeline keeps running. This provides an additional layer of resilience:

# fragment
@classmethod
def handle_error(cls, exc: Exception, message):
    try:
        # Error handling logic that might itself fail
        ...
    except Exception as error_exc:
        # The engine will catch and log this secondary exception
        logger.error(f"Error handler failed: {error_exc}")
        # Processing continues regardless

---

See also

Concept overview: Event Handlers: How event handlers consume and react to domain events.

Related guides:

• Subscribers: For consuming messages from external brokers instead of the internal event store.
• Projectors: For maintaining read-optimized projections from domain events.

Patterns:

• Calling External Systems from Handlers: Splitting a method that persists from one that calls an external system, and the recipe for syncing several facets of one aggregate from one external snapshot.
• Idempotent Event Handlers: Ensuring handlers produce correct results even with duplicate delivery.
• Thin Handlers, Rich Domain: Keeping handlers thin by delegating to domain logic.
• Testing Event-Driven Flows: Strategies for testing event handlers with pytest.