Commands

CQRS ES

Why not call a method on an aggregate directly? In simple cases you can, that's the DDD path with Application Services. But as systems grow, you need to decouple who requests a change from how the change is executed. Commands give you that separation: they can be serialized, routed, processed asynchronously, retried, and audited, all without the caller knowing the details.

Commands represent actions or operations that change the state of the system. A command carries the intent to perform a specific task and the data the task needs. Command handlers process commands and uphold business rules and invariants.

For background on what commands are and how they fit the architecture, see Commands concept.

Defining Commands

A command is defined with the Domain.command decorator:

from datetime import UTC, datetime

from protean import Domain
from protean.fields import DateTime, Identifier

publishing = Domain(name="Publishing")


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


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


@publishing.aggregate
class Article:
    article_id: Identifier(required=True)
    published_at: DateTime(default=utc_now)

A command is always associated with an aggregate class with the part_of option, as seen in the example above. You can use a class reference or a string forward reference (part_of="Article") when the aggregate is defined later in the same module.

Decorator options

Option	Type	Default	Description
`part_of`	class or string	required	The aggregate or process manager this command targets.
`abstract`	bool	`False`	When `True`, the command serves as a base class and cannot be instantiated directly.
`aggregate_cluster`	class	derived from `part_of`	Explicitly set the aggregate cluster for stream routing. Rarely needed.

Submitting Commands

A Protean domain constructs the object graph of all domain elements and routes each command to its Command Handler. You don't need to find the handler yourself.

You can submit the command to the domain for processing with domain.process():

In [1]: command = PublishArticle(article_id="1")

In [2]: publishing.process(command)

`domain.process()` parameters

Parameter	Type	Default	Description
`command`	command instance	required	The command to process.
`asynchronous`	`bool` or `None`	`None`	Override the domain's default processing mode. `None` uses the `command_processing` config. `False` always handles the command at once. When `command_processing` is `"sync"`, `True` handles a command with a registered handler at once outside a Unit of Work, and after the Unit of Work commits inside one, returning `None`.
`idempotency_key`	`str` or `None`	`None`	Caller-provided key for submission-level deduplication (requires Redis).
`raise_on_duplicate`	`bool`	`False`	When `True`, raises `DuplicateCommandError` on duplicate idempotency keys instead of silently returning the cached result.
`priority`	`int` or `None`	`None`	Processing priority for events produced by this command. When priority lanes are enabled, events below the threshold are routed to a backfill stream.
`correlation_id`	`str` or `None`	`None`	Correlation ID for distributed tracing. Propagated to all commands and events in the causal chain. Auto-generated if not provided.
`deadline`	`datetime` or `None`	`None`	Absolute time after which the command must not execute. Stored in metadata and propagated to downstream commands. Mutually exclusive with `timeout`. See Deadlines and Timeouts.
`timeout`	`timedelta` or `None`	`None`	Relative deadline (`now + timeout`), converted to an absolute deadline at submission. Mutually exclusive with `deadline`.

Internally, domain.process() enriches the command with metadata, writes it to the event store, and (for synchronous processing) dispatches it to the appropriate command handler immediately.

Advanced

The following sections cover idempotency and advanced command patterns.

Idempotency

In distributed systems, the same command can arrive more than once due to network retries, broker redelivery, user double-clicks, or saga retries. Protean provides built-in support for command idempotency, ensuring that processing the same command multiple times produces the same effect as processing it once.

The examples from here on use this Order aggregate and its commands:

@domain.aggregate
class Order:
    items: List(content_type=String)


@domain.command(part_of=Order)
class PlaceOrder:
    order_id: Identifier(required=True)
    items: List(content_type=String, required=True)


@domain.command(part_of=Order)
class ReserveStock:
    order_id: Identifier(required=True)


@domain.command(part_of=Order)
class ChargeCard:
    order_id: Identifier(required=True)

Idempotency Keys

When submitting a command, you can provide an idempotency key, a unique token that identifies the specific request:

def place_order(order_id, items):
    return domain.process(
        PlaceOrder(order_id=order_id, items=items),
        idempotency_key="req-abc-123",
    )

If the same idempotency key is submitted again, domain.process() returns the cached result from the first processing without invoking the handler again. Retries are therefore safe. A client can re-send a request due to a timeout and get the same response, whether or not the original succeeded.

The idempotency key is stored in the command's metadata headers, not in the command payload. As a result, the command's payload holds only domain data:

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder):
        # In a handler, the key is accessible via metadata
        key = command._metadata.headers.idempotency_key

        order = Order(id=command.order_id, items=command.items)
        current_domain.repository_for(Order).add(order)
        return key

Note

Idempotency keys are caller-provided, following the model established by Stripe and other well-designed APIs. Protean does not auto-derive keys from command data, only the caller knows whether a submission is a retry or a new intent.

Duplicate Behavior

By default, Protean silently acknowledges a duplicate submission. The caller receives the same result as the first submission. When explicit feedback is needed, use raise_on_duplicate=True:

from protean.exceptions import DuplicateCommandError


def place_order_once(order_id, items):
    try:
        domain.process(
            PlaceOrder(order_id=order_id, items=items),
            idempotency_key="req-abc-123",
            raise_on_duplicate=True,
        )
    except DuplicateCommandError as exc:
        original_result = exc.original_result
        return original_result

Requirements

Submission-level deduplication requires a Redis instance. Configure the connection in domain.toml:

[idempotency]
redis_url = "redis://localhost:6379/5"
ttl = 86400       # Success entries: 24 hours (default)
error_ttl = 60    # Error entries: 60 seconds (default)

Without Redis configured, domain.process() works normally, no deduplication occurs, and no errors are raised.

For a comprehensive treatment of idempotency patterns, including subscription-level deduplication and handler-level strategies, see the Command Idempotency pattern guide.

Synchronous vs Asynchronous Processing

Commands in Protean can be processed either synchronously or asynchronously:

Synchronous processing: The command is processed immediately by the command handler when domain.process() is called. The execution flow is blocked until the command is fully processed.
Asynchronous processing: The command is stored in the event store and later processed by a background worker. The domain.process() call returns immediately without waiting for the command to be fully processed.

You can control the processing mode in two ways:

1. Using the `asynchronous` parameter

When submitting a command, you can explicitly specify whether it should be processed synchronously or asynchronously:

def process_now(command):
    # Process synchronously (default is based on domain configuration)
    return domain.process(command, asynchronous=False)


def process_later(command):
    # Process asynchronously
    return domain.process(command, asynchronous=True)

2. Domain Configuration

You can set the default processing mode for all commands in the domain configuration:

In domain.toml:

command_processing = "sync"  # or "async"

In code:

# Configure default command processing as synchronous
domain.config["command_processing"] = "sync"  # or "async"

By default, Protean sets command_processing to async in the domain configuration.

With command_processing set to "sync", Protean itself handles every command that has a registered handler, even one submitted with asynchronous=True. Such a command is handled at once, unless a Unit of Work is active. Inside a handler, a @use_case method or a with UnitOfWork(): block, process stores a command with a registered handler and returns None. The command runs after the caller commits. A command with no registered handler goes to the engine, and process returns its event store position. See Calling external systems from handlers. Passing asynchronous=False handles a command at once in an "async" domain.

When to use each mode

Synchronous processing is useful when:
You need immediate feedback from the command execution
You want to ensure the command was processed successfully before continuing
The operation is part of a transaction that needs to be completed atomically
Asynchronous processing is beneficial when:
You want to improve UI responsiveness by not blocking the execution flow
The command processing might take a long time
You want to distribute load across background workers
You're implementing CQRS with event sourcing patterns

How Asynchronous Processing Works

Asynchronous commands processing in Protean uses a server/engine component that:

Creates subscriptions for command handlers to listen to their respective command streams
Polls the event store for new commands that haven't been processed yet
Dispatches those commands to the appropriate handlers

To run the Protean server for processing asynchronous commands, use the CLI:

protean server --domain path/to/domain.py

See CLI documentation for more details about the server command and other available CLI options.

The server continually polls the event store for new commands that have the asynchronous flag set to True in their metadata. When found, it dispatches them to the appropriate handlers, keeping track of processed commands to avoid duplicate processing.

Deadlines and Timeouts

A command represents an intent that may only be valid for a limited window. A "charge this card before the checkout session expires" command should not execute an hour later after a queue backlog drains. Protean lets you attach a deadline to a command; if the deadline has passed by the time a handler is about to run, the command is not executed.

Setting a deadline

Pass either an absolute deadline or a relative timeout to domain.process() (they are mutually exclusive):

from datetime import UTC, datetime, timedelta


def charge_by_deadline():
    # Absolute deadline
    return domain.process(
        ChargeCard(order_id="ord-42"),
        deadline=datetime.now(UTC) + timedelta(seconds=30),
    )


def charge_within_timeout():
    # Relative timeout, converted to an absolute deadline at submission,
    # so it survives queue delays
    return domain.process(ChargeCard(order_id="ord-42"), timeout=timedelta(seconds=30))

The deadline is stored on the command's metadata headers (not the payload), and is readable in a handler:

@domain.command_handler(part_of=Order)
class PaymentCommandHandler:
    @handle(ChargeCard)
    def charge_card(self, command: ChargeCard):
        deadline = command._metadata.headers.deadline
        # Charge the card here. This example hands the deadline back.
        return deadline

What happens when a command expires

The behavior differs by processing mode, a deliberate asymmetry:

	Synchronous (`asynchronous=False`)	Asynchronous (default)
Contract	"Fail fast, caller decides"	"Don't run stale work"
On expiry	`domain.process()` raises `CommandExpiredError` to the caller, before the command is written to the event store	The command was already stored and acknowledged; the engine skips it (read position advances, so it is not retried) and emits a `handler.skipped` trace
Who observes the loss	The caller, immediately	Operators, via logs/traces/metrics

from protean.exceptions import CommandExpiredError


def process_before(cmd, past_deadline):
    try:
        domain.process(cmd, asynchronous=False, deadline=past_deadline)
    except CommandExpiredError as exc:
        # The expired command's type string, and the deadline it missed
        return exc.command_type, exc.deadline

An expired command changes no state, no aggregate is loaded, no invariant is evaluated, no event is raised. Expiry is a delivery-layer policy, not a domain-rule violation. The risk it introduces is lost intent: in the async case, the caller is no longer waiting for the result. Protean therefore reports expiries through the protean.command.expired metric (labelled by command_type) and the handler.skipped trace. If dropped commands matter to your business, alert on that metric.

Propagation through the causal chain

When a handler dispatches a downstream command, that command inherits the deadline of the message currently being processed. The whole causal chain is bound by the original deadline unless a downstream call overrides it:

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place(self, command: PlaceOrder):
        # ReserveStock inherits PlaceOrder's deadline automatically
        current_domain.process(ReserveStock(order_id=command.order_id))

Default deadlines

To cover commands without an explicit deadline, you can configure a default validity window instead of passing one on every call. Protean resolves the deadline from these sources in order, and the highest one wins:

Explicit deadline/timeout on domain.process()
Deadline inherited from the parent message in the causal chain
The handling command handler's timeout option
The domain-level command_default_timeout config
No deadline (the default, commands never expire)

# Per-handler default (seconds or a timedelta)
@domain.command_handler(part_of=Order, timeout=30)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place(self, command: PlaceOrder):
        # A PlaceOrder sent with no deadline gets one 30 seconds out
        return command._metadata.headers.deadline

# Domain-wide default (seconds); None disables it
command_default_timeout = 30

Both defaults are disabled by default, so existing applications are unaffected. A default deadline bounds staleness and does nothing about failure. To recover from failures, pair it with retry/backoff and a dead-letter strategy.

For the design rationale (why deadlines are opt-in, the sync/async trade-off, and how this fits DDD) see the Expiring Stale Commands pattern.

Workflow

Command objects are often instantiated by the API controller, which acts as the entry point for external requests into the system. When a client makes a request to the API, the controller receives this request and translates the incoming data into the appropriate command object.

In Protean, the API controller submits the command to the domain object, which then dispatches the command to the appropriate command handler. The Command Handlers section explains how the domain identifies the command handler.

The workflow differs slightly depending on whether synchronous or asynchronous processing is used:

Synchronous Command Flow

Sequence diagram

Asynchronous Command Flow

Sequence diagram

Immutability

Like Events, Commands in Protean are immutable. This means that once a command is created, it cannot be changed.

In [1]: from datetime import datetime, timedelta

In [2]: publish_article_command = PublishArticle(article_id="1")

In [3]: publish_article_command
Out[3]: <PublishArticle: PublishArticle object ({'article_id': '1', 'published_at': '2024-05-28 17:47:35.570857+00:00'})>

In [4]: publish_article_command.published_at = datetime.now() - timedelta(hours=24)
...
IncorrectUsageError: 'Command Objects are immutable and cannot be modified once created'
}

Relationship with Event Processing

Protean offers a similar configuration option for events: the event_processing domain configuration setting.

Protean processes events and commands with the same patterns, so you can build consistent, predictable workflows. You can configure both to suit your specific domain needs:

# Domain-wide configuration
domain.config["command_processing"] = "sync"  # or "async"
domain.config["event_processing"] = "async"  # or "sync"


# Per-command control
def place_now(command):
    # Handle this command at once, even in an "async" domain
    return domain.process(command, asynchronous=False)

With both settings, you can implement architectural patterns like CQRS, Event Sourcing, and Workflow-driven architectures within your Protean applications.

Commands

Defining Commands

Decorator options

Submitting Commands

domain.process() parameters

Idempotency

Idempotency Keys

Duplicate Behavior

Requirements

Synchronous vs Asynchronous Processing

1. Using the asynchronous parameter

2. Domain Configuration

When to use each mode

How Asynchronous Processing Works

Deadlines and Timeouts

Setting a deadline

What happens when a command expires

Propagation through the causal chain

Default deadlines

Workflow

Synchronous Command Flow

Asynchronous Command Flow

Immutability

Relationship with Event Processing

`domain.process()` parameters

1. Using the `asynchronous` parameter