Configuration¶

omni-box is configured per component — there is no single global settings object. The main knobs:

OmniBoxDomainService¶

Constructor flags (all optional, defaults sourced from omni_box.core.constants):

• max_attempts — default retry budget for new events.
• scheduled_at_skew_seconds / scheduled_at_max_future_seconds — validation window for scheduled_at.
• payload_max_bytes — refuse oversized payloads.
• headers_max_count, header_key_max_length, header_value_max_length — header sanity limits.
• last_error_max_length, truncation_suffix — how mark_event_failed truncates long error strings.

OutboxPublisher¶

OutboxPublisher(
    repo,
    broker,
    metrics=None,
    publish_timeout=DEFAULT_PUBLISH_TIMEOUT_SECONDS,
    concurrency_limit=None,
    retry_backoff_base_seconds=DEFAULT_BACKOFF_BASE_SECONDS,   # 10.0
    retry_backoff_cap_seconds=DEFAULT_BACKOFF_CAP_SECONDS,     # 3600.0
)

• publish_timeout — hard timeout for a single broker.publish call. It is treated as the broker's problem, not the row's: it costs no attempt and ends the cycle's publishing. Keep an eye on how it compares with the adapter's own patience — AIOKafkaProducer(request_timeout_ms=...) defaults to 40 s against this timeout's 30 s, so with the defaults the step gives up before aiokafka has said anything, and the log shows a publish timeout where a broker error would be more informative. Set request_timeout_ms below publish_timeout, or raise publish_timeout above it.
• concurrency_limit — wraps publish_batch in an asyncio.Semaphore.
• retry_backoff_base_seconds / retry_backoff_cap_seconds — when a counted failure is retried: after base * 2 ** attempts_made seconds, capped, with equal jitter. 0 retries on the next cycle. The three create_*_processor factories take the same two.

InboxConsumerRunner¶

Key parameters:

• ack_strategy — AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE_INBOX.
• commit_offset_policy — ON_PERSIST (default) or ON_SUCCESS. Only used by AT_LEAST_ONCE.
• exactly_once_commit_on_failed — when True, also commit on handler failure under EXACTLY_ONCE_INBOX (typically combined with DLQ).
• process_timeout — per-message handler timeout.
• concurrency_limit — outer semaphore around process_one.

EventProcessorBuilder¶

• with_lease_ttl(seconds) — lock TTL forwarded to DistributedLockingFetchStrategy.
• with_job_name(name) — used in log records and metrics labels.
• with_metrics(metrics) — wires ProcessingMetrics.
• with_fetch_strategy / with_commit_strategy — override defaults (auto-picked from repository capabilities).

Factories¶

create_outbox_processor, create_inbox_processor, create_dispatching_processor all accept the same observability / safety options:

• metrics
• dlq_storage
• enable_otel
• enable_circuit_breaker, circuit_breaker_failure_threshold, circuit_breaker_recovery_timeout
• additional_steps_before, additional_steps_after — splice extra steps into the pipeline.

Optional pydantic-settings integration¶

With the settings extra installed, omni_box.contrib.settings exposes ready-to-use BaseSettings classes for outbox/inbox runners. Inspect that module for the exact field set — these are convenience helpers, not a required entry point.

DI integration (Dishka)¶

With the dishka extra installed, omni_box.contrib.dishka provides ready-made Provider classes. A minimal example lives in docs/examples/dishka_integration.py.

EventDispatcherProvider provides the EventRouter and the DI-aware router around it. PrometheusInboxMetricsProvider(prefix=None) and PrometheusOutboxMetricsProvider(prefix=None) provide the Prometheus collectors under the keys the rest of the kit takes as metrics=: InboxMetrics | None and OutboxMetrics | None. Each reads the section's own settings — BaseInboxSettings / BaseOutboxSettings, which your application registers under exactly those keys — and provides the collector when observability.enable_metrics is on (the default), None otherwise; the factories, OutboxPublisher and InboxConsumerRunner swap in the no-op on None. The collectors come from get_inbox_metrics / get_outbox_metrics, so a container rebuilt per test does not re-register the series. They need the settings extra, and the metrics extra once metrics are enabled — the collector raises ImportError naming it otherwise.

from dishka import Provider, Scope, make_async_container, provide

from omni_box import EventPublisher, OutboxEventRepository, OutboxMetrics, OutboxPublisher
from omni_box.contrib.dishka import EventDispatcherProvider, PrometheusOutboxMetricsProvider
from omni_box.contrib.settings import BaseOutboxSettings


class AppProvider(Provider):
    scope = Scope.APP

    @provide
    def outbox_settings(self) -> BaseOutboxSettings:
        return BaseOutboxSettings()

    @provide
    def publisher(self, repo: OutboxEventRepository, broker: EventPublisher, metrics: OutboxMetrics | None) -> OutboxPublisher:
        return OutboxPublisher(repo, broker, metrics=metrics)


container = make_async_container(
    EventDispatcherProvider(),
    PrometheusOutboxMetricsProvider(prefix="billing"),
    AppProvider(),
)

A service with only an inbox registers PrometheusInboxMetricsProvider and BaseInboxSettings alone; the two providers are independent.