Logging

Protean ships with structured logging that configures itself. Call domain.init() and every log line — from framework internals, your handler code, and the SQLAlchemy adapter — flows through the same structlog pipeline as JSON in production and colored console output in development.

This guide covers the tasks operators and application developers perform most often. For the full schema of every framework event and every config key, see the Logging reference. For the design rationale behind the wide event pattern, see Logging concepts.

Quick start

from protean import Domain

domain = Domain()
domain.init()  # auto-configures logging

That is the whole setup. Domain.init() auto-detects PROTEAN_ENV, picks a sensible level and format, and installs correlation injection so every log record is queryable by correlation_id. When telemetry.enabled = true, Protean additionally injects OpenTelemetry trace_id, span_id, and trace_flags so logs line up with traces in your APM tool.

To log from application code:

from protean.utils.logging import get_logger

logger = get_logger(__name__)
logger.info("order_placed", order_id="ord-123", total=99.95)

Keyword arguments become structured fields in JSON output and colored key-value pairs in console output. Prefer keyword arguments over f-strings so values remain queryable downstream — see why structured logs? for the rationale.

What a wide event looks like

When a command handler runs, Protean emits one wide event on the protean.access logger. Under PROTEAN_ENV=production the renderer is JSON:

{
  "event": "access.handler_completed",
  "level": "info",
  "logger": "protean.access",
  "timestamp": "2026-04-23T10:15:32.418912Z",
  "kind": "command",
  "message_type": "PlaceOrder",
  "aggregate": "Order",
  "aggregate_id": "ord-9b1c",
  "events_raised": ["OrderPlaced"],
  "events_raised_count": 1,
  "repo_operations": {"loads": 0, "saves": 1},
  "uow_outcome": "committed",
  "handler": "PlaceOrderHandler.handle_place_order",
  "duration_ms": 14.27,
  "status": "ok",
  "correlation_id": "req-abc-123",
  "causation_id": ""
}

A failing handler lifts the level to error, flips status to "failed", and preserves the traceback:

{
  "event": "access.handler_failed",
  "level": "error",
  "logger": "protean.access",
  "kind": "command",
  "message_type": "ChargeCard",
  "handler": "ChargeCardHandler.handle_charge_card",
  "duration_ms": 842.13,
  "status": "failed",
  "error_type": "PaymentDeclined",
  "error_message": "Insufficient funds",
  "correlation_id": "req-def-456",
  "exception": "Traceback (most recent call last):\n  ..."
}

Running under PROTEAN_ENV=development swaps the JSON renderer for a colored ConsoleRenderer, with the same fields inline as key=value pairs. See the Logging reference for the full field list.

Configure via `domain.toml`

The [logging] section is the declarative control surface. A typical production configuration:

[logging]
level = "INFO"
format = "json"
log_dir = "/var/log/myapp"
redact = ["x-internal-token"]

[logging.per_logger]
"myapp.orders" = "DEBUG"

Every key is optional. See the reference page for the full key list, types, defaults, and precedence rules.

Override from the CLI

Every protean command accepts three global flags that take precedence over domain.toml:

protean server --log-level DEBUG
protean server --log-format json
protean server --log-config ./logging.json      # full dictConfig JSON

--log-config bypasses the environment-aware setup and applies the supplied JSON via logging.config.dictConfig(). The correlation filter is still installed on the root logger afterwards.

protean server --debug is deprecated in favor of --log-level DEBUG and will be removed in v0.17.0.

Override programmatically

When domain.toml is not the right shape — tests, one-off scripts, embedded domains — call Domain.configure_logging() directly. Explicit keyword arguments override domain.toml but still read PROTEAN_LOG_LEVEL as an override for level unless level= is passed:

domain.configure_logging(level="DEBUG", format="json")

If you already called domain.init(), calling configure_logging() again replaces the handlers and re-installs the correlation filter. This is safe to do in tests to reset state.

Enrich wide events with business context

Protean emits one wide event per handled command, event, query, or projector on the protean.access logger. The framework fills in domain context automatically; application code adds business-specific fields with bind_event_context():

from protean import handle
from protean.utils.logging import bind_event_context

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place(self, command: PlaceOrder) -> None:
        bind_event_context(
            user_tier=command.user_tier,
            order_total=float(command.total),
            coupon_applied=command.coupon_code is not None,
        )
        # ... handler logic ...

The framework and application fields merge into the single wide event emitted when the handler returns. See the reference for field-reservation rules and the concept page for guidance on choosing queryable dimensions.

Use structured events in application code

get_logger() returns a structlog logger bound to the stdlib logger of the given name. Events are keyword arguments, not f-strings:

from protean.utils.logging import get_logger

logger = get_logger(__name__)
logger.info("payment_refunded", order_id="ord-123", amount=19.99, reason="customer_request")

For context that should appear on every record inside a scope, use add_context():

from protean.utils.logging import add_context, clear_context

add_context(request_id="abc-123", tenant_id="tenant-42")
try:
    logger.info("processing")          # includes request_id and tenant_id
    logger.info("processed")
finally:
    clear_context()

add_context() uses contextvars, so it propagates correctly across await boundaries and thread-local scope.

Control wide event volume with tail sampling

By default Protean emits one wide event per handled message. At scale — millions of messages per day — this can become expensive to store and query. Tail sampling keeps every error and slow request (the events that actually help you debug) and samples the happy path at a configurable rate.

Enable it declaratively:

[logging.sampling]
enabled = true
default_rate = 0.05         # keep 5 % of happy-path events
always_keep_errors = true   # status="failed" / ERROR+ level
always_keep_slow = true     # status="slow"
critical_streams = ["Payment*", "Auth*"]   # fnmatch globs on message_type

Every kept event carries three metadata fields so log aggregators can compute accurate throughput from sampled data:

{
  "event": "access.handler_completed",
  "sampling_decision": "kept",
  "sampling_rule": "random",
  "sampling_rate": 0.05
}

To reconstruct the true count: actual_count = sampled_count / sampling_rate.

Rules apply in order; first match wins: always_keep_errors → always_keep_slow → critical_streams → random sampling at default_rate. See the reference for per-key details and the concept page for why tail sampling beats head sampling.

Keep the safety defaults on

Setting always_keep_errors = false or always_keep_slow = false combined with a low default_rate will silently drop the events you most want to see. The safety defaults exist for a reason.

Emit a security event

protean.security is a dedicated channel for invariant, validation, and authorization failures that cross a domain boundary. Framework code emits to this channel automatically for aggregate invariant violations and the three Invalid* exceptions. To emit from application code:

from protean.integrations.logging import (
    SECURITY_EVENT_VALIDATION_FAILED,
    log_security_event,
)

def check_admin_access(user, resource):
    if not user.can_access(resource):
        log_security_event(
            SECURITY_EVENT_VALIDATION_FAILED,
            aggregate="Resource",
            aggregate_id=resource.id,
            user_id=user.id,
            reason="not_authorized",
        )
        raise PermissionDenied()

correlation_id and causation_id are auto-injected from the active domain context. Route this logger to your SIEM with no sampling or format filters attached so every entry is delivered intact. See the reference for the full list of framework-emitted event types.

Disable auto-configuration

When Protean is embedded inside an application that already configured its own logging (Django, a custom server, an OS-level journald shim), set PROTEAN_NO_AUTO_LOGGING=1 before calling domain.init():

export PROTEAN_NO_AUTO_LOGGING=1

You can then wire whichever parts of Protean's integration you want manually:

import logging
from protean.integrations.logging import ProteanCorrelationFilter

logging.getLogger().addFilter(ProteanCorrelationFilter())

Domain.init() also detects a pre-configured root logger (handlers already attached) and skips its auto-configuration in that case, so in many embedded setups no env var is needed.

Minimize noise in tests

In conftest.py:

from protean.utils.logging import configure_for_testing

configure_for_testing()

This sets the root logger to WARNING and removes file handlers. Tests that assert on log output with pytest's caplog fixture still work because structlog writes to stdlib handlers.