Command Handlers

CQRS ES

Commands carry intent, but someone needs to act on them. Command handlers carry out the work a command asks for. They receive a command, load the right aggregate, call the appropriate domain method, and persist the result. They keep your aggregates free of infrastructure concerns and your API layer free of domain logic.

For background on how command handlers fit the CQRS architecture, see Command Handlers concept.

Defining a Command Handler

Command Handlers are defined with the Domain.command_handler decorator:

from datetime import UTC, datetime
from enum import Enum

from protean import Domain, handle
from protean.fields import DateTime, Identifier, String
from protean.utils.globals import current_domain

publishing = Domain(name="Publishing")


def utc_now():
    return datetime.now(UTC)


class ArticleStatus(Enum):
    DRAFT = "DRAFT"
    PUBLISHED = "PUBLISHED"


@publishing.command(part_of="Article")
class PublishArticle:
    article_id: Identifier(required=True)
    published_at: DateTime(default=utc_now)


@publishing.event(part_of="Article")
class ArticlePublished:
    article_id: Identifier(required=True)
    published_at: DateTime()


@publishing.aggregate
class Article:
    article_id: Identifier(identifier=True)
    status: String(choices=ArticleStatus, default=ArticleStatus.DRAFT.value)
    published_at: DateTime(default=utc_now)

    def publish(self, published_at: datetime) -> None:
        self.status = ArticleStatus.PUBLISHED.value
        self.published_at = published_at

        self.raise_(
            ArticlePublished(article_id=self.article_id, published_at=published_at)
        )


@publishing.command_handler(part_of=Article)
class ArticleCommandHandler:
    @handle(PublishArticle)
    def publish_article(self, command: PublishArticle) -> None:
        article = current_domain.repository_for(Article).get(command.article_id)
        article.publish(published_at=command.published_at)
        current_domain.repository_for(Article).add(article)

Decorator options

Option	Type	Default	Description
`part_of`	class or string	required	The aggregate this handler processes commands for.
`stream_category`	`str`	derived from `part_of`	Read-only. Always derived from the aggregate's stream category. Cannot be overridden.
`timeout`	`int`/`float` seconds or `timedelta`	`None`	Default validity window for commands routed to this handler. Overrides the domain-level `command_default_timeout`; overridden by an explicit `deadline`/`timeout` on `domain.process()`. See Deadlines and Timeouts.
`retries`	`int`	`None`	Max auto-retry attempts on transient exceptions. Overrides the domain-level `server.transient_retry`; `None` defers to it. See Retrying transient failures.
`backoff`	`str`	`None`	Backoff strategy between retries: `"exponential"`, `"linear"`, or `"fixed"`. `None` defers to config.
`retry_exceptions`	exception types or dotted-path strings	`None`	Which exceptions count as transient. `None` uses the configured default (`ConnectionError`, `TimeoutError`, `SendError`).
`sequential_by`	`str`	`None`	Name of an attribute on every command this handler handles. Registration checks that every handled command has the attribute. Commands do not go through the outbox yet, so they are not split by this value. See Sequential processing for what the option does on a command handler today.
`subscription_type`	`SubscriptionType` or `str`	`None`	Subscription type for the server's subscription, as the enum member or its name, such as `"stream"`. `None` uses the default.
`subscription_profile`	`SubscriptionProfile` or `str`	`None`	Subscription profile preset, as the enum member or a profile name. A name can be a built-in profile or a custom profile. `None` uses the default.
`subscription_config`	`dict`	`{}`	Custom subscription settings that override the type and profile defaults.
`suppress_checks`	tuple of `str`	`()`	Diagnostic codes `protean check` should not report for this handler.

The `@handle` decorator

Each method that processes a command is decorated with @handle, imported from protean:

from protean import handle

@handle(CommandClass) registers the method as the handler for that command type. It also wraps the method body in a Unit of Work, providing automatic transaction management. A single command can only be handled by one handler method across the entire domain. Protean enforces one-handler-per-command.

How routing works

When you call domain.process(command), Protean uses the command's part_of aggregate to find the registered command handler for that aggregate. It then looks up the @handle-decorated method that matches the command's type and invokes it. You never need to wire this routing manually. It is all derived from the part_of associations on the command and command handler.

Stream Category

Command handlers automatically subscribe to commands in their associated aggregate's stream category. When a command is processed, it is routed to the appropriate handler based on the command's target aggregate and its stream category.

For example, if an Order aggregate has a stream category of order, its command handler will listen for commands on the order:command stream category. Commands are stored in streams following the pattern <domain>::<stream_category>-<aggregate_id>.

Unlike event handlers, command handlers cannot override their stream category. The stream category is always derived from the part_of aggregate. As a result, Protean always routes commands to the single handler responsible for the target aggregate.

Learn more about stream categories and message routing in the Stream Categories guide.

Internal workflow

Sequence diagram

Application Submits Command: The application calls domain.process(command), which enriches the command with metadata and writes it to the event store.
Command is Stored: Protean persists the command to the event store before any processing occurs, so an audit trail exists even if the handler fails.
Domain Dispatches to Handler: The domain identifies the correct command handler based on the command's part_of aggregate and dispatches it.
Command Handler Loads Aggregate: The command handler loads the target aggregate from the repository using current_domain.repository_for().
Aggregate Mutates: The handler invokes the appropriate method on the aggregate, which applies business rules, validates invariants, and changes its internal state.
Command Handler Persists Aggregate: The handler persists the modified aggregate back to the repository, committing changes within the Unit of Work.

Return Values from Command Handlers

Command handlers can optionally return values to the caller when processed synchronously. This behavior is determined by how the command is processed by the domain.

Synchronous Processing

When commands are processed synchronously, the command handler's return value is passed back to the caller. This is useful for:

Returning newly created resource identifiers
Providing validation or processing results
Returning calculated values or status information

To process a command synchronously and receive its return value:

def register_account(command):
    # Process command synchronously and get the return value
    result = domain.process(command, asynchronous=False)
    return result

Example of a command handler that returns a value:

@domain.aggregate
class Account:
    email: String(required=True)
    name: String(required=True)


@domain.command(part_of=Account)
class RegisterCommand:
    email: String(required=True)
    name: String(required=True)


@domain.command_handler(part_of=Account)
class AccountCommandHandler:
    @handle(RegisterCommand)
    def register(self, command: RegisterCommand) -> str:
        account = Account(email=command.email, name=command.name)
        current_domain.repository_for(Account).add(account)

        # Return the account ID for immediate use
        return account.id

Asynchronous Processing

When commands are processed asynchronously (the default behavior), the command handler's return value is not passed back to the caller. Instead, the domain's process method returns the position of the command in the event store:

def submit_registration(command):
    # Process command asynchronously (default)
    position = domain.process(command)  # or domain.process(command, asynchronous=True)
    return position

In asynchronous processing, commands are handled in the background by the Protean Engine, and any return values from the command handler are ignored.

Configuring Default Processing Behavior

The default command processing behavior can be configured in the domain's configuration:

# ...
command_processing = "sync"  # or "async"
# ...

When set to "sync", every command is handled at once, except one submitted with asynchronous=True while a Unit of Work is active. That command runs after the Unit of Work commits, in its own Unit of Work. When set to "async", commands are queued for the server unless submitted with asynchronous=False.

Idempotency in Handlers

When a command is submitted with an idempotency key (via domain.process(command, idempotency_key="...")), the key is available in the handler through the command's metadata:

@domain.command(part_of=Account)
class ChargeCard:
    account_id: Identifier(required=True)
    amount: Float(required=True)


@domain.command_handler(part_of=Account)
class AccountCommandHandler:
    @handle(ChargeCard)
    def charge(self, command: ChargeCard):
        key = command._metadata.headers.idempotency_key

        # Pass through to external APIs that support idempotency
        stripe.PaymentIntent.create(
            amount=command.amount,
            currency="usd",
            idempotency_key=key,
        )

This is useful for:

Pass-through to external APIs: Many services (Stripe, payment processors) accept idempotency keys natively. Passing the same key ensures end-to-end retry safety.
Handler-level deduplication: For additive operations (incrementing counters, adding items), the handler can check the key against a set of previously processed keys stored on the aggregate.

For detailed patterns including natural idempotency, check-then-act, and event-sourced aggregate strategies, see the Command Idempotency pattern guide.

Unit of Work

Each command handler method runs within a UnitOfWork context. If the method completes successfully, Protean commits all changes. If the method raises an exception, Protean rolls everything back.

For details on how the Unit of Work pattern works, see the Unit of Work guide.

Note

A UnitOfWork context applies to objects in the aggregate cluster, and not multiple aggregates. A Command Handler method can load multiple aggregates to perform the business process, but should never persist more than one at a time. Other aggregates should be synced eventually through domain events.

Error Handling

Error handling differs between synchronous and asynchronous command processing:

Synchronous (asynchronous=False): Exceptions raised in the handler propagate directly to the caller (the code that called domain.process()). The UoW is rolled back automatically. There is no handle_error hook. The caller is responsible for catching and handling the exception.
Asynchronous (default): The Protean Engine catches exceptions during background processing and invokes the handler's handle_error class method (if defined). The engine then continues processing the next command. Since there is no caller waiting for a response, the handle_error hook is the only mechanism for custom error recovery.

The `handle_error` Method

You can define a handle_error class method in your command handler to handle exceptions:

@domain.command_handler(part_of=Account)
class AccountCommandHandler:
    @handle(RegisterCommand)
    def register(self, command: RegisterCommand):
        # Command handling logic that might raise exceptions
        ...

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

        # Perform recovery operations
        # Example: notify monitoring systems, attempt retry, etc.

How It Works

When an exception occurs in a command handler method, the Protean Engine catches it.
The engine logs detailed error information including stack traces.
The engine calls the command handler's handle_error method, passing:
The original exception that was raised
The command message being processed when the exception occurred
After handle_error returns, processing continues with the next command.

Handling Errors in the Error Handler

If an exception occurs within the handle_error method itself, the Protean Engine will catch that exception too, log it, and continue processing. So even a failure in error handling does not crash the system.

# fragment
@classmethod
def handle_error(cls, exc: Exception, message):
    try:
        # Potentially risky error handling logic
        ...
    except Exception as error_exc:
        # This secondary exception will be caught by the engine
        logger.error(f"Error in error handler: {error_exc}")
        # The engine will continue processing regardless

Best Practices

Keep error handlers simple, and avoid logic that might itself fail.
Use error handlers for logging, notification, and simple recovery.
Don't throw exceptions from error handlers unless absolutely necessary.
Consider enabling transient-failure retries for recoverable infrastructure errors.

Retrying transient failures

Handlers can automatically retry when they fail with a transient infrastructure error, such as a dropped connection or a timeout. Enable it per handler with retries:

@domain.command(part_of=Account)
class DebitAccount:
    account_id: Identifier(required=True)
    amount: Float(required=True)


@domain.command_handler(part_of=Account, retries=3, backoff="exponential")
class AccountCommandHandler:
    @handle(DebitAccount)
    def debit(self, command: DebitAccount):
        # A ConnectionError here is retried up to 3 times with
        # exponential backoff before propagating.
        ...

Each attempt opens its own Unit of Work, so a failed attempt is rolled back cleanly before the next one. Some handlers run inside an outer Unit of Work, such as another handler or a use case that calls domain.process() synchronously. Such a handler does not retry and ignores its own retries and retry_exceptions. Its exception propagates to the outermost handler, which re-runs the whole chain in a fresh Unit of Work when its own policy covers the exception. Otherwise the exception reaches the caller. Protean retries only the exceptions that count as transient, which by default are ConnectionError, TimeoutError, and SendError. Narrow or widen the set with retry_exceptions:

@domain.command_handler(part_of=Account, retries=2, retry_exceptions=[ConnectionError])
class AccountCommandHandler:
    @handle(DebitAccount)
    def debit(self, command: DebitAccount):
        # Only a ConnectionError is retried. A TimeoutError propagates at once.
        ...

Transient retry is separate from the version-conflict (OCC) retry for ExpectedVersionError. The two combine, each with its own attempt budget. Transient retry is opt-in and disabled unless retries is set or server.transient_retry is enabled. The same options work on @domain.event_handler. See server.transient_retry for the domain-wide defaults.

Testing Command Handlers

The simplest way to test a command handler is to submit a command synchronously and verify the resulting state:

# fragment
def test_publish_article(test_domain):
    # Arrange
    article = Article(article_id="1", status="DRAFT")
    test_domain.repository_for(Article).add(article)

    # Act
    test_domain.process(
        PublishArticle(article_id="1"),
        asynchronous=False,
    )

    # Assert
    refreshed = test_domain.repository_for(Article).get("1")
    assert refreshed.status == "PUBLISHED"

Key points:

Use asynchronous=False to process the command synchronously in tests, so the handler runs immediately and you can assert on the result.
Configure command_processing = "sync" in your test domain config to make this the default for all tests.
You can also test return values directly: result = test_domain.process(cmd, asynchronous=False).