Configuration

Applies to: DDD · CQRS · Event Sourcing

Protean's configuration is managed through the Domain object, which can be configured in multiple ways:

Direct Configuration

Pass a configuration dictionary when initializing the Domain object:

domain = Domain(config={'debug': True, 'testing': True})

Configuration Files

Place configuration can be supplied in a TOML file in your project directory or up to two levels of parent directories.

Protean searches for configuration files in the following order:

.domain.toml
domain.toml
pyproject.toml (under the [tool.protean] section)

Generating a new configuration file

When initializing a new Protean application using the new command, a domain.toml configuration file is automatically generated with sensible defaults.

A sample configuration file is below:

debug = true
testing = true
secret_key = "tvTpk3PAfkGr5x9!2sFU%XpW7bR8cwKA"
identity_strategy = "uuid"
identity_type = "string"
event_processing = "sync"
command_processing = "sync"

[databases.default]
provider = "memory"

[databases.memory]
provider = "memory"

[brokers.default]
provider = "inline"

[caches.default]
provider = "memory"

[event_store]
provider = "memory"

[custom]
foo = "bar"

[staging]
event_processing = "async"
command_processing = "sync"

[staging.databases.default]
provider = "sqlite"
database_url = "sqlite:///test.db"

[staging.brokers.default]
provider = "redis_pubsub"
URI = "redis://staging.example.com:6379/2"
TTL = 300

[staging.custom]
foo = "qux"

[prod]
event_processing = "async"
command_processing = "async"

[prod.databases.default]
provider = "postgresql"
database_url = "postgresql://postgres:postgres@localhost:5432/postgres"

[prod.brokers.default]
provider = "redis_pubsub"
URI = "redis://prod.example.com:6379/2"
TTL = 30

[prod.event_store]
provider = "message_db"
database_uri = "postgresql://message_store@localhost:5433/message_store"

[prod.custom]
foo = "quux"

[prod.telemetry]
enabled = true
service_name = "my-service"
exporter = "otlp"
endpoint = "http://otel-collector:4317"

Basic Configuration Parameters

`debug`

Specifies if the application is running in debug mode.

Do not enable debug mode when deploying in production.

Default: False

`testing`

Enable testing mode. Exceptions are propagated rather than handled by the domain’s error handlers. Extensions may also change their behavior to facilitate easier testing. You should enable this in your own tests.

Default: False

`secret_key`

A secret key that will be used for security related needs by extensions or your application. It should be a long random bytes or str.

You can generate a secret key with the following command:

> python -c 'import secrets; print(secrets.token_hex())'
c4bf0121035265bf44657217c33a7d041fe9e505961fc7da5d976aa0eaf5cf94

Do not reveal your secret key when posting questions or committing code.

`env`

Reserved for informational use. Protean does not read this key to select which environment overlay ([staging], [prod], etc.) to apply: overlay selection is driven entirely by the PROTEAN_ENV environment variable, described in Multiple Environments. Setting env in domain.toml has no effect on Protean's own behavior.

Default: None

`identity_strategy`

The default strategy to use to generate an identity value. Can be overridden at the Auto field level.

Supported options are uuid and function.

If the identity_strategy is chosen to be a function, identity_function has to be mandatorily specified during domain object initialization.

Default: uuid

`identity_type`

The type of the identity value. Can be overridden at the Auto field level.

Supported options are integer, string, and uuid.

Default: string

`command_processing`

Whether to process commands synchronously or asynchronously.

Supported options are sync and async.

Default: async

Note

The protean new scaffold writes an explicit command_processing = "sync" override into the generated domain.toml (see the sample above), which is why new projects behave synchronously by default. The framework default, absent any override, is async.

`command_default_timeout`

Default validity window, in seconds, applied to every command when no explicit deadline/timeout is passed to domain.process() and the handling command handler declares no timeout option. When the window elapses before a handler runs, the command is rejected (sync) or skipped without retry (async). See Deadlines and Timeouts.

None disables the default: commands never expire unless a deadline is requested explicitly or through a handler's timeout option.

Default: None

`event_processing`

Whether to process events synchronously or asynchronously.

Supported options are sync and async.

Default: async

Note

The protean new scaffold writes an explicit event_processing = "sync" override into the generated domain.toml (see the sample above), which is why new projects behave synchronously by default. The framework default, absent any override, is async.

`message_processing`

Whether to process incoming messages from external brokers (subscribers) synchronously or asynchronously.

Supported options are sync and async.

Default: async

`source_uri`

Overrides the CloudEvents source URI-reference attribute stamped on outgoing event/command envelopes. When unset, Protean derives urn:protean:<domain-name> from the domain's name.

Default: None

`lenient_deserialization`

When true, deserializing a stored event or command payload drops fields that no longer exist on the current class (recording their names under metadata.extensions["_dropped_fields"]) instead of raising a DeserializationError. Useful when reading legacy payloads written before a field was removed, without an upcaster.

The dropped-field record is read-time observability on the deserialized message: it is not written back to the event store and, being _-prefixed, is not emitted as a CloudEvents extension. Dropping only removes unknown fields, a payload still missing a required field raises as usual.

The default is strict: an unknown field raises, so a typo or genuine schema drift is not silently swallowed. A per-event lenient meta option (@domain.event(lenient=True/False)) overrides this config for a specific class. Field-rename renamed_from aliases resolve first, so a renamed old key is kept, not dropped.

Default: false

`snapshot_threshold`

The threshold number of aggregate events after which a snapshot is created to optimize performance.

Applies only when aggregates are event sourced.

Default: 10

Adapter Configuration

`databases`

Database repositories are used to access the underlying data store for persisting and retrieving domain objects.

They are defined in the [databases] section.

[databases.default]
provider = "memory"

[databases.memory]
provider = "memory"

[databases.sqlite]
provider = "sqlite"
database_uri = "sqlite:///test.db"

You can define as many databases as you need. The default database is identified by the default key, and is used when you do not specify a database name when accessing the domain.

The only other database defined by default is memory, which is the in-memory stub database provider.

The persistence store defined here is then specified in the provider key of aggregates and entities to assign them a specific database.

@domain.aggregate(provider="sqlite")  # (1)
class User:
    name: String(max_length=50)
    email: String(max_length=254)

sqlite is the key of the database definition in the [databases.sqlite] section.

SQLAlchemy providers (postgresql, mssql) accept pool tuning keys directly on the [databases.<name>] block:

Key	Default	Purpose
`pool_size`	`5`	Base connections kept open per worker
`max_overflow`	`10`	Additional temporary connections beyond `pool_size`
`pool_recycle`	unset	Recycle connections older than N seconds

Setting pool_size below 5 triggers a LOW_POOL_SIZE warning from protean check (suppressed when PROTEAN_ENV is development or testing).

Read more in Adapters → Database or the full catalogue in Server Hardening reference.

`caches`

This section holds definitions for cache infrastructure.

[caches.default]
provider = "memory"

[caches.redis]
provider = "redis"
URI = "redis://127.0.0.1:6379/2"
TTL = 300
max_connections = 50
socket_timeout = 5
socket_connect_timeout = 5
retry_on_timeout = true

Default provider: memory

Redis caches forward max_connections, socket_timeout, socket_connect_timeout, and retry_on_timeout to the underlying redis.ConnectionPool.

Read more in Adapters → Cache section.

`broker`

This section holds configurations for message brokers.

[brokers.default]
provider = "memory"

[brokers.redis]
provider = "redis_pubsub"
URI = "redis://127.0.0.1:6379/0"
IS_ASYNC = true
max_connections = 50
socket_timeout = 5
socket_connect_timeout = 5
retry_on_timeout = true

Default provider: memory

Redis brokers (redis_pubsub, redis_stream) forward the same pool keys as the Redis cache to redis.ConnectionPool.

Read more in Adapters → Broker section.

`event_store`

The event store that stores event and command messages is defined in this section.

[event_store]
provider = "message_db"
database_uri = "postgresql://message_store@localhost:5433/message_store"
max_connections = 20

Note that there can only be only event store defined per domain.

Default provider: memory

MessageDB forwards max_connections through conn_info to cap the connection pool used for event reads and writes.

Read more in Adapters → Event Store section.

`server`

This section configures the Protean server (async message processing engine). It controls subscription behavior, outbox processing, and handler-specific settings.

[server]
# Default subscription settings
default_subscription_type = "stream"      # "stream" or "event_store"
default_subscription_profile = "production"  # Profile for defaults
messages_per_tick = 100                   # Messages per processing cycle

# StreamSubscription defaults
[server.stream_subscription]
blocking_timeout_ms = 5000    # Blocking read timeout
max_retries = 3               # Retries before DLQ
retry_delay_seconds = 1       # Delay between retries
enable_dlq = true             # Enable dead letter queue

# EventStoreSubscription defaults
[server.event_store_subscription]
position_update_interval = 10  # Messages between position writes
max_retries = 3                # Retries before marking exhausted
retry_delay_seconds = 1        # Delay between recovery retries
enable_recovery = true         # Enable periodic recovery pass
recovery_interval_seconds = 30 # Interval between recovery sweeps
gap_timeout_seconds = 5        # $all subs: hold at a global_position gap this long before abandoning it

# BrokerSubscription defaults
[server.broker_subscription]
max_retries = 3               # Retries before DLQ
retry_delay_seconds = 1       # Delay between retries
enable_dlq = true             # Enable dead letter queue

# Version conflict auto-retry
# Retries ExpectedVersionError at the handler level before
# the error reaches the subscription retry/DLQ pipeline
[server.version_retry]
enabled = true               # Enable/disable auto-retry
max_retries = 3              # Fast retries before propagating
base_delay_seconds = 0.05    # 50ms initial backoff
max_delay_seconds = 1.0      # Cap backoff at 1 second

# Transient-failure auto-retry (distinct from version_retry)
# Retries handlers that fail with transient infrastructure errors
# before the error reaches the subscription retry/DLQ pipeline.
# Opt-in: disabled by default. Per-handler retries=/backoff=/
# retry_exceptions= options override these.
[server.transient_retry]
enabled = false              # Off by default; turn on to retry transient errors
max_retries = 3              # Retries before propagating
backoff = "exponential"      # exponential | linear | fixed
base_delay_seconds = 0.1     # Initial backoff delay
max_delay_seconds = 5.0      # Cap backoff at 5 seconds
# Only genuinely transient exceptions belong here. Dotted paths or
# bare builtin names.
exceptions = [
    "builtins.ConnectionError",
    "builtins.TimeoutError",
    "protean.exceptions.SendError",
]

# DLQ maintenance (retention + alerting)
# Off by default; opt in by setting enabled = true.
[server.dlq]
enabled = false              # Master switch for the maintenance task
retention_hours = 168        # Trim DLQ entries older than this (default: 7 days)
alert_threshold = 100        # Warn when a DLQ stream reaches this depth
check_interval_seconds = 60  # How often the maintenance cycle runs
alert_callback = "myapp.alerts.notify_oncall"  # Optional dotted path, called on breach

# Kubernetes-compatible health HTTP server
# Enabled by default on port 8080; disable for tests or embedded use.
[server.health]
enabled = true               # Start the built-in health server
host = "127.0.0.1"           # Bind address (loopback; use "0.0.0.0" to expose)
port = 8080                  # Listen port
port_auto_increment = false  # Try 8081, 8082, ... if the port is taken

# Two-lane event/command routing (backfill vs. primary stream)
# Off by default; opt in by setting enabled = true.
[server.priority_lanes]
enabled = false               # Off by default; enable two-lane routing
threshold = 0                 # Priority below this routes to the backfill lane
backfill_suffix = "backfill"  # Suffix for the backfill stream/DLQ

# Handler-specific overrides
[server.subscriptions.OrderEventHandler]
profile = "fast"
messages_per_tick = 50
# Per-subscription DLQ overrides (inherit from [server.dlq] when unset)
dlq_retention_hours = 72
dlq_alert_threshold = 25

[server.subscriptions.InventoryProjector]
subscription_type = "event_store"
profile = "projection"

Subscription Profiles

Pre-configured settings for common scenarios:

Profile	Type	Use Case
`production`	stream	High throughput with reliability
`fast`	stream	Low-latency processing
`batch`	stream	High-volume batch processing
`debug`	stream	Development and debugging
`projection`	event_store	Building read models

Read more in Server → Configuration section.

DLQ Maintenance

The [server.dlq] section configures a periodic maintenance task that trims old DLQ entries and alerts when DLQ streams grow deep. It is disabled by default, set enabled = true to activate it. Requires a broker that implements the DLQ contract (currently Redis Streams).

Key	Type	Default	Description
`enabled`	bool	`false`	Master switch. When `true`, the Engine starts the maintenance task.
`retention_hours`	int	`168`	Trim DLQ entries older than this window (default 7 days).
`alert_threshold`	int	`100`	Emit a `WARNING` log and invoke `alert_callback` when a DLQ stream reaches this depth.
`check_interval_seconds`	int	`60`	How often the maintenance cycle runs.
`alert_callback`	str \| None	`None`	Dotted import path to a callable invoked on threshold breach. Signature: `(dlq_stream: str, depth: int, threshold: int) -> None`. Exceptions raised by the callback are caught and logged.

Per-subscription overrides are available on each handler's [server.subscriptions.<HandlerName>] block:

Key	Type	Default	Description
`dlq_retention_hours`	int \| None	inherit	Override `retention_hours` for this subscription's DLQ stream only.
`dlq_alert_threshold`	int \| None	inherit	Override `alert_threshold` for this subscription's DLQ stream only.

For the operational workflow (discover, inspect, replay, purge) see Dead Letter Queues.

Health Checks

The [server.health] section configures the built-in HTTP server used for Kubernetes liveness and readiness probes. The server is enabled by default on port 8080.

Key	Type	Default	Description
`enabled`	bool	`true`	Start the health HTTP server. Set to `false` for tests or embedded deployments.
`host`	str	`"127.0.0.1"`	Bind address. Defaults to loopback so probes are not exposed off-host; set `"0.0.0.0"` for out-of-pod probes.
`port`	int	`8080`	Listen port.
`port_auto_increment`	bool	`false`	When `true`, if `port` is already bound the server walks up (`8081`, `8082`, ...) until it finds a free one. Lets several engines share a host without colliding.

The server exposes GET /healthz, GET /livez, and GET /readyz. For the probe response bodies, readiness semantics, and FastAPI router factory, see Server Hardening reference.

Priority Lanes

The [server.priority_lanes] section configures two-lane event/command routing: production messages flow through the primary stream while bulk/migration messages are routed to a separate backfill stream. It is disabled by default: set enabled = true to activate it.

Key	Type	Default	Description
`enabled`	bool	`false`	Whether to activate priority lanes. When `false`, all messages use a single stream regardless of priority.
`threshold`	int	`0`	Priority values strictly below this threshold are routed to the backfill lane.
`backfill_suffix`	str	`"backfill"`	Suffix appended to the stream category to form the backfill stream name.

Read more in Priority Lanes and Using Priority Lanes.

`outbox`

This section configures the transactional outbox pattern for reliable message delivery. The outbox is automatically enabled when server.default_subscription_type is set to "stream". The legacy enable_outbox = true flag still works for backward compatibility.

[server]
default_subscription_type = "stream"  # Enables outbox automatically

[outbox]
broker = "default"         # Target broker for publishing
messages_per_tick = 10     # Messages per processing cycle
tick_interval = 1          # Seconds between cycles
max_tick_interval = 30     # Adaptive backoff cap when idle (omit to disable backoff)

# Retry configuration
[outbox.retry]
max_attempts = 3           # Maximum retry attempts
base_delay_seconds = 60    # Initial retry delay
max_backoff_seconds = 3600 # Maximum retry delay
backoff_multiplier = 2     # Exponential backoff multiplier
jitter = true              # Add randomization to delays

# Cleanup configuration
[outbox.cleanup]
published_retention_hours = 168   # Keep published for 7 days
abandoned_retention_hours = 720   # Keep abandoned for 30 days
batch_size = 5000                 # Rows deleted per bounded cleanup batch

Cleanup runs as bounded batch_size deletes rather than one unbounded DELETE, so large backlogs are cleared without long lock holds; each batch commits before the next when cleanup runs standalone.

Read more in Server → Outbox Pattern section.

`telemetry`

This section configures OpenTelemetry distributed tracing and metrics.

[telemetry]
enabled = true
service_name = "my-service"
exporter = "otlp"
endpoint = "http://localhost:4317"

[telemetry.resource_attributes]
"deployment.environment" = "production"

Key	Type	Default	Description
`enabled`	bool	`false`	Master switch for all OTel instrumentation
`service_name`	string	domain name	Populates the OTel `service.name` resource attribute
`exporter`	string	`"otlp"`	Span and metric exporter: `"otlp"` or `"console"`
`endpoint`	string	SDK default	OTLP collector endpoint (gRPC)
`resource_attributes`	table	`{}`	Additional OTel resource attributes

When disabled (the default), all tracing and metrics are no-ops with zero overhead. Requires pip install "protean[telemetry]".

`idempotency`

This section configures command idempotency deduplication. When configured with a Redis instance, domain.process() can detect and deduplicate repeated command submissions using caller-provided idempotency keys.

[idempotency]
redis_url = "redis://localhost:6379/5"  # Redis connection URL
ttl = 86400                              # Success entry TTL in seconds (default: 24 hours)
error_ttl = 60                           # Error entry TTL in seconds (default: 60s)

Key	Description	Default
`redis_url`	Redis connection URL for the idempotency cache. When `None`, deduplication is disabled.	`None`
`ttl`	Time-to-live for successful idempotency entries (seconds). After expiry, the same key can be reused.	`86400` (24 hours)
`error_ttl`	Time-to-live for error entries (seconds). Short TTL allows retries after transient failures.	`60`

Without redis_url configured, idempotency deduplication is disabled and domain.process() works normally with no errors.

`consume_idempotency`

This section configures cleanup of the consume-side idempotency markers used by projectors to detect and skip already-processed messages (the ProcessedMessage marker; distinct from the command idempotency store above). See ADR-0017.

[consume_idempotency.cleanup]
retention_hours = 168  # Prune markers older than this (default: 7 days)
batch_size = 5000      # Rows deleted per bounded cleanup batch

Key	Description	Default
`cleanup.retention_hours`	Prune `ProcessedMessage` markers older than this window (hours).	`168` (7 days)
`cleanup.batch_size`	Rows deleted per bounded cleanup batch.	`5000`

Read more in Projectors and the protean data idempotency CLI command.

`logging`

This section configures Protean's structured logging: level, output format, file rotation, slow-handler/slow-query thresholds, and redaction of sensitive fields.

[logging]
level = "INFO"
format = "auto"

Full key reference, redaction rules, and tail sampling: Logging.

`lint`

This section configures the diagnostics reported by protean check. All keys are optional.

[lint]
level = "warn"                        # exit-code severity floor
aggregate_size_limit = 5              # entities before AGGREGATE_TOO_LARGE
handler_breadth_limit = 5             # message types before HANDLER_TOO_BROAD
check_infra_imports = false           # flag domain modules importing protean.adapters
check_adapter_calls = false           # flag domain elements calling adapters directly
rules = ["my_app.lint.check_names"]   # dotted paths to custom rule callables

[lint.suppressions]
UNHANDLED_EVENT = 3                    # grandfather the first 3 findings

Key	Type	Default	Description
`level`	str	`"warn"`	Exit-code severity floor for `protean check`. `"error"` fails only on errors; `"warn"` fails on errors and warnings; `"info"` fails on any error, warning, or info finding. Errors always exit `1`; a gating non-error finding exits `2`.
`aggregate_size_limit`	int	`5`	Entity count above which an aggregate emits `AGGREGATE_TOO_LARGE`.
`handler_breadth_limit`	int	`5`	Message-type count above which a handler emits `HANDLER_TOO_BROAD`.
`check_infra_imports`	bool	`false`	When `true`, `protean check` AST-parses each resolvable domain element's source module (skipping elements with no file-backed module and imports guarded by `TYPE_CHECKING`, `try`/`except`, or a function body) and emits `INFRA_IMPORT_IN_DOMAIN` for any top-level import from `protean.adapters`. Off by default because it reads source files.
`check_adapter_calls`	bool	`false`	When `true`, `protean check` emits `ADAPTER_CALL_IN_DOMAIN` for a domain element calling into an adapter directly rather than through a port. Off by default because it reads source files.
`rules`	list[str]	`[]`	Dotted import paths to custom lint callables with the signature `(ir: dict) -> list[dict]`.
`suppressions`	table	`{}`	A `{CODE: N}` map that grandfathers the first `N` findings of each code, in a deterministic `(code, element, field, message)` order. Findings beyond `N` are still reported.

level sets the CI failure floor without hiding findings; the --level CLI flag only filters what is displayed and never changes the exit code.

Individual elements can silence specific codes for themselves with the suppress_checks decorator option (for example @domain.aggregate(suppress_checks=["AGGREGATE_TOO_LARGE"])), which takes precedence over the [lint.suppressions] allow-list.

`[server.partitioning]`

Tunables for sequential_by partition ownership. Defaults suit most deployments; raise the lease TTL only if your handlers routinely pause longer than it.

[server.partitioning]
lease_ttl_ms = 15000              # how long an ownership lease survives without renewal
heartbeat_interval_seconds = 3.0  # how often the owner renews it
poll_interval_seconds = 0.25      # how often a worker looks for unowned partitions
reap_idle_ms = 3600000            # idle time before a cold partition is pruned

Key	Type	Default	Purpose
`lease_ttl_ms`	int	`15000`	How long a partition's ownership lease survives without renewal. A worker that dies frees its partitions after this.
`heartbeat_interval_seconds`	float	`3.0`	How often the owner renews the lease. Keep it well under `lease_ttl_ms`.
`poll_interval_seconds`	float	`0.25`	How often a worker looks for partitions it could take.
`reap_idle_ms`	int	`3600000`	How long a partition must be empty and idle before it is pruned from the index.

See Sequential processing for the model these configure, and ADR-0028 for why ownership is leased and fenced.

Custom Attributes

Custom attributes can be defined in toml under the [custom] section (or [tool.protean.custom] if you are using the pyproject.toml file).

Custom attributes are also made available as domain attributes.

debug = true
testing = true

[custom]
FOO = "bar"

In [1]: domain = Domain()

In [2]: domain.config["custom"]["FOO"]
Out[2]: 'bar'

In [3]: domain.FOO
Out[3]: 'bar'

Multiple Environments

Most applications need more than one configuration. At minimum, there should be separate configurations for production and for local development. The toml configuration file can hold configurations for different environments.

The current environment is gathered from an environment variable named PROTEAN_ENV.

The string specified in PROTEAN_ENV is used as a qualifier in the configuration.

[databases.default]
provider = "memory"

[staging.databases.default]
provider = "sqlite"
database_url = "sqlite:///test.db"

[prod.databases.default]
provider = "postgresql"
database_url = "postgresql://postgres:postgres@localhost:5432/postgres"

Protean has a default configuration with memory stubs that is overridden by configurations in the toml file, which can further be over-ridden by an environment-specific configuration, as seen above. There are two environment specific settings above for databases - an sqlite db configuration for staging and a postgresql db configuration for prod.

Testing Overlays

A common use of environment overlays is dual-mode testing, running the same test suite against in-memory adapters for speed and real infrastructure for confidence. Define a [test] overlay (the default for pytest) and a [memory] overlay, then switch from the command line:

# Base configuration (development)
[databases.default]
provider = "postgresql"
database_uri = "postgresql://localhost/myapp_local"

[brokers.default]
provider = "redis"
URI = "redis://localhost:6379/0"

# Test overlay: separate database, sync processing
[test]
testing = true
event_processing = "sync"

[test.databases.default]
database_uri = "postgresql://localhost/myapp_test"

# Memory overlay: all in-memory adapters, no Docker needed
[memory]
testing = true
event_processing = "sync"

[memory.databases.default]
provider = "memory"

[memory.event_store]
provider = "memory"

[memory.brokers.default]
provider = "inline"

pytest --protean-env memory   # Fast — in-memory, no infrastructure
pytest                        # Thorough — real adapters (PROTEAN_ENV=test)

Protean's pytest plugin sets PROTEAN_ENV before test collection via the --protean-env option (default: test). See the Dual-Mode Testing pattern for the full approach.