event-handler

---
name: event-handler
description: Define a Protean event handler, a class that consumes domain events raised by aggregates and runs side effects such as syncing state across aggregates, sending notifications, or triggering downstream processes. Event handlers are associated with an aggregate via part_of or with a stream via stream_category, and use the @handle decorator to process specific event types. They are fire-and-forget and return no values. Also covers cross-aggregate synchronization via events, keeping one aggregate per transaction. Use when you need to react to a domain event, handle an event, process events from another aggregate, sync state between aggregates, implement eventual consistency, add notifications or side effects for domain events, or when the user asks to "create an event handler", "handle an event", "react to an event", "sync aggregates", "add a listener for events", "update another aggregate when X happens", "add cross-aggregate coordination", "when order ships reduce inventory", or "add an event-driven sync".
license: Apache-2.0
compatibility: Requires Python 3.11+, protean framework
metadata:
  author: proteanhq
  version: "0.1"
  category: element
  diagnostic_codes:
    - EVENT_HANDLER_FOREIGN_EVENT
    - HANDLER_TOO_BROAD
    - HANDLER_PERSISTS_AND_CALLS_OUT
---

# Event Handler

An event handler runs side effects when an aggregate raises an event: it updates its own aggregate, or hands a change to another aggregate with a command. This skill defines handlers, their options, and the cross-aggregate pattern.

## Basic structure

```python
from protean import Domain, handle
from protean.fields import Identifier, String

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

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

@domain.aggregate
class Order:
    order_id: Identifier(identifier=True)
    status: String(default="draft")
    confirmation_number: String()

    def place(self):
        self.status = "placed"
        self.raise_(OrderPlaced(order_id=self.order_id))

@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = domain.repository_for(Order)
        order = repo.get(event.order_id)
        order.confirmation_number = f"CONF-{event.order_id}"
        repo.add(order)
```

## Key rules

1. **part_of or stream_category required**: Handler must be associated with an aggregate or stream: `@domain.event_handler(part_of=Order)`
2. **Use @handle decorator**: Each handler method is decorated with `@handle(EventClass)` or `@handle("$any")`
3. **Handler methods take self and event**: Signature: `def method_name(self, event: EventClass)`
4. **No return values**: Event handlers do NOT return values (CQRS pattern, fire-and-forget)
5. **Implicit UnitOfWork**: `@handle` wraps every handler invocation in its own Unit of Work, which is one real database transaction on a provider with transactions, such as the SQL databases. A handler method that reaches no repository opens no transaction. On a version conflict (`ExpectedVersionError`) the whole method runs again: with the default `max_retries=3` it runs up to 4 times in total, with waits that double between runs. Each run opens its own Unit of Work. A handler called while an outer Unit of Work is active joins it and does not retry: the conflict propagates to the caller. Configure it under `server.version_retry`
6. **Multiple handlers per event**: Unlike commands, multiple event handlers can process the same event
7. **Import handle from protean**: `from protean import handle` (not from protean.core)
8. **part_of uses class reference**: Handler uses `part_of=AggregateClass` by convention. A string reference (`part_of="AggregateName"`) also works and resolves at `init`, so the aggregate may be defined after the handler
9. **Hand off to another aggregate with a command**: To change aggregate B when aggregate A raises an event, put the handler in A's cluster (`part_of=A`, no `stream_category`) and have it call `current_domain.process(...)` with a command that belongs to B. B's command handler does the write. A handler in B's cluster that reacts to A's event is what `check` reports as `EVENT_HANDLER_FOREIGN_EVENT`

## Handler options

| Option | Default | Description |
|--------|---------|-------------|
| `part_of` | `None` | The aggregate whose cluster the handler belongs to, as a class or a string name. Registration fails unless `part_of` or `stream_category` is set |
| `source_stream` | `None` | Handle only events whose origin stream is in this stream category. An event's origin stream is the stream of the event that led to the command that raised it. Applies to event-store subscriptions |
| `stream_category` | derived from part_of | The stream category the handler listens to. Defaults to the `part_of` aggregate's stream category; to name another, use `Aggregate.meta_.stream_category` |
| `retries` | `None` | Maximum retries when the handler fails with a transient exception. Overrides the domain's `server.transient_retry` setting; `None` uses that setting |
| `backoff` | `None` | Delay strategy between transient retries: `"exponential"`, `"linear"` or `"fixed"`. `None` uses the domain setting |
| `retry_exceptions` | `None` | Exception types treated as transient, as classes or dotted paths. `None` uses the domain setting |
| `sequential_by` | `None` | Name of an attribute on every event this handler handles. Events with the same value are processed one at a time. `None` turns this off |
| `subscription_type` | `None` | `SubscriptionType` for the server's subscription (`STREAM` or `EVENT_STORE`). `None` uses the default |
| `subscription_profile` | `None` | `SubscriptionProfile` preset for the subscription (`PRODUCTION`, `FAST`, `BATCH`, `DEBUG`, `PROJECTION`). `None` uses the default |
| `subscription_config` | `{}` | Custom subscription settings, such as `messages_per_tick`, that override the type and profile defaults |
| `suppress_checks` | `()` | Diagnostic codes `protean check` should not report for this handler |

## Quick example: Multiple events

The example uses these elements:

```python
@domain.aggregate
class Account:
    name: String(required=True)
    status: String(default="active")

@domain.event(part_of=Account)
class AccountRegistered:
    account_id: Identifier(required=True)
    name: String(required=True)

@domain.event(part_of=Account)
class AccountSuspended:
    account_id: Identifier(required=True)
    reason: String(required=True)

@domain.aggregate
class Notification:
    notification_id: Identifier(identifier=True)
    message: String(required=True)

@domain.command(part_of=Notification)
class SendNotification:
    notification_id: Identifier(required=True)
    message: String(required=True)
```

Each method issues a `SendNotification` command. An account can be suspended
more than once, so the account id cannot name one notification. The
notification id is the event's message id (`event._metadata.headers.id`),
which stays the same when the event is delivered again. See the caveat in
"Making the receiving end safe to repeat" below.

```python
@domain.event_handler(part_of=Account)
class AccountNotifier:
    @handle(AccountRegistered)
    def on_registered(self, event: AccountRegistered):
        current_domain.process(
            SendNotification(
                notification_id=event._metadata.headers.id,
                message=f"Welcome {event.name}!",
            )
        )

    @handle(AccountSuspended)
    def on_suspended(self, event: AccountSuspended):
        current_domain.process(
            SendNotification(
                notification_id=event._metadata.headers.id,
                message=f"Account suspended: {event.reason}",
            )
        )
```

## Quick example: Catch-all ($any) handler

`TaskAuditor` records every event on `Task`'s stream as an `AuditLog` entry:

```python
@domain.aggregate
class Task:
    title: String(required=True)

@domain.aggregate
class AuditLog:
    event_type: String(required=True)

@domain.event_handler(part_of=AuditLog, stream_category=Task.meta_.stream_category)
class TaskAuditor:
    @handle("$any")
    def on_any_event(self, event):
        audit = AuditLog(event_type=event.__class__.__name__)
        domain.repository_for(AuditLog).add(audit)
```

## Error handling

Override `handle_error` classmethod for custom error recovery during async processing:

```python
import logging

logger = logging.getLogger(__name__)

@domain.aggregate
class Shipment:
    order_id: Identifier(required=True)

@domain.event(part_of=Shipment)
class ShipmentDispatched:
    shipment_id: Identifier(required=True)

@domain.event_handler(part_of=Shipment)
class ShipmentNotifier:
    @handle(ShipmentDispatched)
    def on_dispatched(self, event):
        logger.info("Shipment %s dispatched", event.shipment_id)

    @classmethod
    def handle_error(cls, exc: Exception, message) -> None:
        logger.error(f"Shipment event failed: {exc}")
```

## Cross-aggregate sync pattern

The primary use case for event handlers is coordinating state changes between aggregates
while maintaining the **one aggregate per transaction** rule.

### The principle

Never modify two aggregates in the same handler. Hand the change to the other aggregate with a command (Key rule 9), and put enough data in the event that the handler does not need to load the source aggregate. With async command processing, the other aggregate's command handler runs in a **separate transaction** (eventual consistency). With sync command processing, as in tests, it runs inside the event handler's unit of work and commits with it.

| Approach | Correct? | Why |
|----------|----------|-----|
| Handler modifies source + target in one call | No | Two aggregates in one transaction |
| Handler in A's cluster issues a command to B | Yes | Each aggregate changes only through its own command handler |
| Handler in B's cluster listens to A's stream and writes B | No | `check` reports `EVENT_HANDLER_FOREIGN_EVENT`: the two clusters are coupled directly |
| Command handler loads and modifies two aggregates | No | Violates consistency boundary |

For a flow with several causally dependent steps, where each step waits on the
outcome of the one before, use a process manager. See
[process-manager](../process-manager/SKILL.md).

### Making the receiving end safe to repeat

Events are delivered at least once, so the same event can reach the handler
twice and issue the same command twice. The command carries a deterministic id
taken from the event, such as the order id. B's command handler returns without
changes when that work is already done:

- **When the command creates B**, give B that id and check for it first:
  `try: repository.get(command.order_id)`, add B on `ObjectNotFoundError`, and
  return otherwise.
- **When the command updates B**, keep a list of applied ids on B (for example
  `applied_order_ids: List(content_type=String)`). The command handler returns
  early when the id is already in the list, and the aggregate method appends it.
  The list grows with every applied id and loads with B each time.

When one event drives several writes to B, derive one id per write, such as
`f"{order_id}:{product_id}"`. Add up the lines for the same product first, so
each product gets one write. Two lines for one product would otherwise share an
id, and the handler would skip the second as a repeat. Never generate a fresh `uuid4()` in the handler,
because a redelivered event would then look like new work.

Prefer an id from the event's payload. When the same kind of event can recur
for one source and the payload has no id for each occurrence, use the event's
message id, `event._metadata.headers.id`. On an aggregate that is not event
sourced it has the form `<stream>-<version>.<n>` and stays the same on every
delivery. It names one change as long as an aggregate id is never reused. Each
committed save writes a new version, so events from separate saves get separate
ids, whether or not the instance is reloaded in between.

Keys like `f"{order_id}:{product_id}"` and message ids are strings, not UUIDs.
A guard that stores them as B's identity needs the default
`identity_type` (`"string"`). With `identity_type="uuid"`, the database
rejects them.

### Common patterns

```
Order.ship()      → OrderShipped     → InventorySyncHandler (part_of=Order)      → ReduceStock          → Inventory.reduce_stock()
Payment.confirm() → PaymentConfirmed → SubscriptionSyncHandler (part_of=Payment) → ActivateSubscription → Subscription.activate()
Task.assign_to()  → TaskAssigned     → WorkloadSyncHandler (part_of=Task)        → RecordAssignment     → TeamMember.record_assignment()
```

### Building a cross-aggregate handler

The full runnable version, including the `ShipOrder` command and `Order`'s
command handler, is [cross_sync_order_inventory.py](assets/cross_sync_order_inventory.py).
A `ShipOrder` command ships the order, `Order` raises `OrderShipped`,
`InventorySyncHandler` in `Order`'s cluster issues `ReduceStock`, and
`InventoryCommandHandler` reduces the stock unless it has already applied that
order. The block starts a new domain, because its `Order` is not the one from
[Basic structure](#basic-structure).

```python
domain = Domain()

# 1. Event on source aggregate
@domain.event(part_of="Order")
class OrderShipped:
    order_id: Identifier(required=True)
    product_id: Identifier(required=True)
    quantity: Integer(required=True)

# 2. Source aggregate raises event (Order's command handler calls ship())
@domain.aggregate
class Order:
    product_id: Identifier(required=True)
    quantity: Integer(required=True)
    status: String(default="placed")

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

# 3. Command to the target aggregate carries the order id from the event
@domain.command(part_of="Inventory")
class ReduceStock:
    order_id: Identifier(required=True)
    product_id: Identifier(required=True)
    quantity: Integer(required=True)

# 4. Target aggregate records which orders it has applied
@domain.aggregate
class Inventory:
    product_id: Identifier(required=True)
    stock_level: Integer(required=True)
    applied_order_ids: List(content_type=String)

    def reduce_stock(self, order_id, quantity):
        self.stock_level -= quantity
        self.applied_order_ids = [*self.applied_order_ids, order_id]

# 5. Target's command handler does the write, and skips an order it has applied
@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(product_id=command.product_id)
        if command.order_id in inventory.applied_order_ids:
            return  # a redelivered OrderShipped; reducing again would double count
        inventory.reduce_stock(command.order_id, command.quantity)
        repo.add(inventory)

# 6. Event handler in the source's cluster hands off with the command
@domain.event_handler(part_of=Order)
class InventorySyncHandler:
    @handle(OrderShipped)
    def on_order_shipped(self, event: OrderShipped):
        current_domain.process(ReduceStock(
            order_id=event.order_id,
            product_id=event.product_id,
            quantity=event.quantity,
        ))
```

## Common mistakes

### Handling another cluster's event (`EVENT_HANDLER_FOREIGN_EVENT`)

```python
# fragment
# Wrong! check reports EVENT_HANDLER_FOREIGN_EVENT
@domain.event_handler(part_of=Inventory, stream_category=Order.meta_.stream_category)
class InventorySyncHandler:
    @handle(OrderShipped)
    def on_order_shipped(self, event: OrderShipped):
        inventory = domain.repository_for(Inventory).find_by(product_id=event.product_id)
        inventory.reduce_stock(event.quantity)
        domain.repository_for(Inventory).add(inventory)
```

The handler belongs to `Inventory` but reacts to `Order`'s event, so the two
clusters are coupled directly. Instead: put the handler in `Order`'s cluster and
issue a `ReduceStock` command, as in
[Building a cross-aggregate handler](#building-a-cross-aggregate-handler).

### Manually wrapping in UnitOfWork

```python
# fragment
@handle(OrderPlaced)
def handle(self, event):
    with UnitOfWork():  # Unnecessary! Already implicit
        order = domain.repository_for(Order).get(event.order_id)
        domain.repository_for(Order).add(order)
```

Instead: Let the implicit UnitOfWork handle it. A Unit of Work started inside the handler joins the one `@handle` already opened. There are no savepoints, and only the outermost Unit of Work commits or rolls back

### Business logic in the handler

```python
# fragment
@handle(OrderPlaced)
def confirm_order(self, event):
    order = domain.repository_for(Order).get(event.order_id)
    if order.status != "placed":  # Business logic leak!
        raise ValueError("Only placed orders can be confirmed")
    order.status = "confirmed"
    domain.repository_for(Order).add(order)
```

Instead: Keep business logic in the aggregate (an `order.confirm()` method), handler only orchestrates

### What `check` reports

`check` inspects your event handlers and reports these diagnostics:

- `EVENT_HANDLER_FOREIGN_EVENT`: the handler reacts to an event owned by another cluster, which couples the two clusters directly. Move the handler into the cluster that owns the event and have it issue a command that belongs to this cluster. For a flow with several causally dependent steps, use a `ProcessManager` that reacts to the source event and issues the command.
- `HANDLER_TOO_BROAD`: the handler handles more message types than the configured `[lint] handler_breadth_limit`, so it has grown into a catch-all. Split it into focused handlers, or raise the limit if the breadth is intentional.
- `HANDLER_PERSISTS_AND_CALLS_OUT`: one handler method calls an external system after its first `repository_for(...)`, so the call runs with the Unit of Work's transaction open, holding row locks and a pooled connection for as long as the call takes, and a retry re-runs the method and re-issues the call. A call made before any repository access runs outside the transaction and is not flagged. Split the method into one that persists and one that calls out; when the call must follow the write, raise an event from the persisting side and handle that instead.

## Detailed references

### Core Concepts
- [Same-Aggregate Handling](references/same-aggregate.md): Handler processes events from its own aggregate
- [Cross-Aggregate Handling](references/cross-aggregate.md): Handler in the event's own cluster changes another aggregate through a command
- [Eventual Consistency](references/eventual-consistency.md): Trade-offs and guarantees
- [Cross-Aggregate Patterns](references/cross-aggregate-patterns.md): Common sync scenarios
- [Error Handling](references/error-handling.md): Custom error recovery with handle_error
- [Catch-All ($any) Handler](references/any-handler.md): Processing any event on the stream
- [Anti-patterns](references/anti-patterns.md): Common mistakes and how to avoid them

### Complete Examples
- [Same-Aggregate Handler](assets/event_handler_same_aggregate.py): Handler for its own aggregate's events
- [Cross-Aggregate Handler](assets/event_handler_cross_aggregate.py): Order's handler sends Inventory a command when an order ships
- [Order-Inventory Sync](assets/cross_sync_order_inventory.py): OrderShipped reduces Inventory stock
- [Payment-Subscription Sync](assets/cross_sync_payment_subscription.py): PaymentConfirmed activates Subscription
- [Multi-Event Sync](assets/cross_sync_multi_event.py): Multiple events from one source trigger different target updates
- [Multiple Events](assets/event_handler_multiple_events.py): Handler with multiple @handle methods
- [Error Handling](assets/event_handler_error_handling.py): Custom handle_error classmethod
- [Catch-All Handler](assets/event_handler_any_event.py): @handle("$any") for all events

### Related Skills
- [event](../event/SKILL.md): Events are the input to event handlers
- [aggregate](../aggregate/SKILL.md): Event handlers are always connected to aggregates
- [command-handler](../command-handler/SKILL.md): Analogous pattern for commands, and the write path a cross-aggregate handler hands off to
- [process-manager](../process-manager/SKILL.md): Flows with several causally dependent steps across aggregates
- [refactor-introduce-events](../refactor-introduce-events/SKILL.md): Refactoring direct calls into event-driven flows

## Verify your work

- [Verify with check](../../references/verify-with-check.md): run `check` and resolve what it reports

• SKILL.md 20 KB
• __init__.py 1 KB
• assets/__init__.py 1 KB
• assets/cross_sync_multi_event.py 9 KB
• assets/cross_sync_order_inventory.py 6 KB
• assets/cross_sync_payment_subscription.py 6 KB
• assets/event_handler_any_event.py 4 KB
• assets/event_handler_cross_aggregate.py 6 KB
• assets/event_handler_error_handling.py 7 KB
• assets/event_handler_multiple_events.py 10 KB
• assets/event_handler_same_aggregate.py 3 KB
• references/anti-patterns.md 7 KB
• references/any-handler.md 3 KB
• references/cross-aggregate-patterns.md 5 KB
• references/cross-aggregate.md 5 KB
• references/error-handling.md 3 KB
• references/eventual-consistency.md 1 KB
• references/same-aggregate.md 2 KB