api_reference.md

API Reference

This page documents the public surface of omni-box — everything re-exported from the top-level package and the most useful submodules. For an auto-generated, source-of-truth listing run:

import omni_box
print(sorted(omni_box.__all__))
print(omni_box.__version__)

Domain models

Imported from omni_box (defined in omni_box.core.models).

• BaseEvent — common fields: id, event_type, payload, headers, status, attempts_made, max_attempts, last_error, timing (created_at, scheduled_at, completed_at), locking (locked_at, locked_by), tracing (trace_id, correlation_id, causation_id, idempotency_key, schema_version). There is no updated_at on the entity — that column lives on the ORM mixin (EventMixin) only.
• OutboxEvent — adds aggregate_type, aggregate_id, topic, partition_key.
• InboxEvent — adds message_id, consumer_group, source.
• BaseEventSchema — Pydantic schema base used by ad-hoc payload models.
• EventStatus (StrEnum) — PENDING = "pending", COMPLETED = "completed", FAILED = "failed".

Domain services

OmniBoxDomainService

Factory for validated event entities and the source of truth for lock/transition rules.

• create_outbox_event(aggregate_type, aggregate_id, event_type, topic, partition_key, payload, *, headers=None, idempotency_key=None, trace_id=None, correlation_id=None, causation_id=None, schema_version=None, max_attempts=None, scheduled_at=None) -> OutboxEvent
• create_inbox_event(message_id, consumer_group, source, event_type, payload, *, headers=None, trace_id=None, correlation_id=None, causation_id=None, schema_version=None) -> InboxEvent
• lock_event(event, worker_id, locked_at), refresh_event_lock, unlock_event, force_unlock_event
• mark_event_completed, mark_event_failed (supports count_as_attempt, next_retry_at)
• is_lock_stale(event, now, stale_timeout_seconds=None) -> bool

OmniBoxMaintenanceService

Operational helpers. Requires a repository that implements SupportsRetentionPolicies.

• release_stale_locks(stale_timeout_seconds) -> int
• cleanup_old_events(retention_days, batch_size=..., max_iterations=...) -> int

Protocols

From omni_box.core.protocols.

• Repositories: EventRepository[T], OutboxEventRepository, InboxEventRepository, FetchFilters, RepositoryCapabilities.
• Capabilities: SupportsBulkOperations[T], SupportsDistributedLocking[T], SupportsRetentionPolicies.
• Broker: EventPublisher, EventConsumer, ConsumedMessage, AckHandle / NullAckHandle, EnvelopeParser, EnvelopeData, InboxHandler.
• Transaction providers: InboxTransactionProviderProtocol, OutboxTransactionProviderProtocol (under omni_box.core.protocols.transaction).
• Metrics: InboxMetrics, OutboxMetrics, ProcessingMetrics (under omni_box.core.protocols.metrics).

Pipeline

From omni_box.core.pipeline.

EventProcessorBuilder[T: BaseEvent]

Fluent builder. Picks DistributedLockingFetchStrategy + BulkCommitStrategy automatically when the repository advertises the matching capabilities; falls back to OptimisticLockingFetchStrategy + SingleCommitStrategy.

Method	Description
add_step(step)	Append a ProcessingStep[T].
with_fetch_strategy(strategy)	Override the auto-picked fetch strategy.
with_commit_strategy(strategy)	Override the auto-picked commit strategy.
with_metrics(metrics)	Wire a ProcessingMetrics collector. Steps reach it as ProcessingContext.metrics.
with_lease_ttl(seconds)	Lock TTL used by DistributedLockingFetchStrategy.
with_job_name(name)	Logging/metrics label.
build()	Returns an EventBatchProcessor[T].

EventBatchProcessor[T: BaseEvent]

• process_batch(worker_id, batch_size, shutdown_requested_func=None, **fetch_filters) -> BatchProcessingResult
• BatchProcessingResult carries processed_event_ids, failed_counted, failed_noncounted, remaining_event_ids, commit_failed.
• shutdown_requested_func is polled before the fetch and before every event. Once it returns True the batch stops: what was already processed is committed, and the events left untouched come back in remaining_event_ids. Asked before the fetch, it locks nothing at all.

Built-in steps (omni_box.core.pipeline.steps)

Step	Purpose	Notes
HandlerExecutionStep	Runs the user handler inside the pipeline with a timeout.	Required terminal step in every processor. A TransientError out of the handler is recorded with count_as_attempt=False and rescheduled; a timeout counts.
PublisherExecutionStep	The outbox's handler step; installed by create_outbox_processor.	Subclasses HandlerExecutionStep. The publish timeout is transient too, and the first transient failure ends the batch's publishing — the remaining events are recorded the same way, unpublished, with their schedule untouched.
SiblingDeduplicationStep	Skips an InboxEvent if a sibling row with the same (message_id, consumer_group) is already completed.	Calls InboxEventRepository.has_completed_sibling_for_inbox_key. A no-op on a non-partitioned table — see storage adapters.
MetricsStep	Pushes batch lifecycle counters into an InboxMetrics / OutboxMetrics sink.
OpenTelemetryStep(service_name=...)	Creates spans for each batch/event.	Requires opentelemetry extra.
CircuitBreakerStep(failure_threshold, recovery_timeout_seconds)	Stops batch processing after consecutive failures.	State is process-local; not distributed. Add Redis-backed coordination if you need cross-replica behaviour.
DLQStep(dlq_storage)	Routes failed events to a DLQStorage[T] after exhausting retries.	Best-effort, non-transactional. Transient (count_as_attempt=False) failures are never routed to DLQ.

Strategies (omni_box.core.pipeline.strategies)

• Fetch: DistributedLockingFetchStrategy, OptimisticLockingFetchStrategy, FilteredFetchStrategy.
• Commit: BulkCommitStrategy, SingleCommitStrategy.

Application-layer services

OutboxPublisher (omni_box.application.services.publish)

OutboxPublisher(
    repo: OutboxEventRepository,
    broker: EventPublisher,
    metrics: OutboxMetrics | None = None,
    publish_timeout: float = DEFAULT_PUBLISH_TIMEOUT_SECONDS,
    concurrency_limit: int | None = None,
)

• publish_batch(worker_id, batch_size, shutdown_requested_func=None, **fetch_filters) — one fetch/lock/publish/write-back cycle through the session behind repo. The library opens no transaction and commits nothing: wrap the call in one of your own. shutdown_requested_func behaves as it does on process_batch above.

InboxConsumerRunner (omni_box.application.services.consume)

InboxConsumerRunner(
    consumer: EventConsumer,
    transaction_provider: InboxTransactionProviderProtocol,
    handler: InboxHandler | None = None,
    *,
    worker_id: str,
    consumer_group: str,
    domain_service: OmniBoxDomainService | None = None,
    ack_strategy: AckStrategy = AckStrategy.EXACTLY_ONCE_INBOX,
    commit_offset_policy: CommitOffsetPolicy = CommitOffsetPolicy.ON_PERSIST,
    exactly_once_commit_on_failed: bool = False,
    process_timeout: float = DEFAULT_PROCESS_TIMEOUT_SECONDS,
    concurrency_limit: int | None = None,
    metrics: InboxMetrics | None = None,
)

• start(), stop()
• run_forever() — loops calling process_one, applies exponential backoff on errors.
• process_one() -> InboxConsumeResult (message_id, event_id, committed, processed, duplicate).

Enums:

• AckStrategy — AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE_INBOX.
• CommitOffsetPolicy — ON_PERSIST, ON_SUCCESS (only used by AT_LEAST_ONCE).

Factories (omni_box.application.factories)

All three return EventBatchProcessor[T].

• create_outbox_processor(repo, publisher, *, publish_timeout, metrics=None, dlq_storage=None, enable_otel=False, enable_circuit_breaker=False, ...)
• create_inbox_processor(repo, handler, *, skip_duplicate_siblings=True, filter_sources=None, process_timeout=..., ...)
• create_dispatching_processor(repo, router, *, dependencies=None, ...) — uses an EventRouter.

Dispatch (omni_box.core.dispatch)

• EventRouter(normalize_topic=None) — registry of handlers keyed by (topic, event_type, schema_version). register_handler(event_type, topic, handler, schema_version=None, handler_name=None) registers a callable; register_class(cls, topic=None) / register_instance(obj, topic=None) sweep a BaseEventHandler for decorated methods. dispatch(event, topic, repo, **dependencies) tries the exact schema_version, then a registered migration, then the version-agnostic entry, and returns a failed EventHandlerResult when nothing matches.
• BaseEventHandler — base class for class-based handlers; set topic on the subclass.
• event_handler(event_type, topic=None, schema_version=None) — decorator. There is no source parameter, and the decorator only marks the method: a router has to register it. create_dispatching_processor dispatches on event.source as the topic.
• create_dispatching_handler(router, **dependencies) — the router-backed handler that create_dispatching_processor installs; use it when you assemble the pipeline yourself.
• DispatchName (str | StrEnum) and as_dispatch_str(name) — the topic / event-type spellings accepted by everything above.

Handler results

From omni_box.core.services.results, re-exported at the top level.

• EventHandlerStatus, EventHandlerResult, BatchProcessingResult.
• Helpers: handler_completed(status=EventHandlerStatus.COMPLETED), handler_retry(message, *, count_as_attempt=True, next_retry_at=None, status=EventHandlerStatus.RETRY), handler_skipped(status=EventHandlerStatus.SKIPPED). All three take a status, not a free-text reason; the message on a retry is the one written to last_error.

Converters (omni_box.core.converters)

• EventConverter (protocol)
• RawEventConverter — body is the raw payload.
• SchemaVersionedConverter — body is {"schema_version": ..., "payload": ...}.
• EnvelopeEventConverter — full envelope with tracing identifiers (re-exported at the top level).

Exceptions

All inherit from OmniBoxError.

• Storage: StorageError, StorageConnectionError, StorageTimeoutError, StorageTransactionError, StorageIntegrityError.
• Domain / locking: EventNotLockedError, EventLockedByAnotherWorkerError, EventAlreadyLockedError, InvalidEventStateError, EventConcurrentUpdateError.
• Transient: TransientError — raised by a publisher or a handler to say the failure belongs to the environment and not to the event. The pipeline records it without spending an attempt and retries the row on the next cycle.
• Misc: UnsupportedCapabilityError, InboxPersistError.

Infrastructure adapters

PostgreSQL (extra: postgres)

omni_box.infra.storage.postgres:

• ORM bases: OutboxEventDBBase, InboxEventDBBase, OutboxEventPartitionedDBBase, InboxEventPartitionedDBBase, plus the underlying EventMixin, OutboxColumnsMixin, InboxColumnsMixin.
• Repositories: PostgresOutboxRepository, PostgresInboxRepository, PostgresEventRepository (shared base). Both expose session — the AsyncSession they were built on, read-only — which is how a handler passed to InboxConsumerRunner writes its side effects in the transaction that inserts the inbox row.
• Helpers: UnConstrainedEnum, get_event_constraints(table_name, include_created_at_in_unique=False).

Kafka (extra: kafka)

omni_box.infra.brokers.kafka:

• KafkaEventPublisher(producer, converter, *, max_infra_retries=3) — built on top of aiokafka.AIOKafkaProducer. The caller owns the producer lifecycle (start/stop). A broker that does not answer — connection and node errors, request and client timeouts, and an unknown topic the broker will not confirm with a metadata refresh either — is retried max_infra_retries times and then raised as TransientError, which costs the row no attempt. Anything else, including a record the broker rejects, is raised as it is and counts.
• KafkaEventConsumer — wraps aiokafka.AIOKafkaConsumer and exposes per-record AckHandles. Use DefaultEnvelopeParser or provide your own EnvelopeParser.

Neither adapter depends on any external "kit" package; only aiokafka is required.

Prometheus (extra: metrics)

omni_box.infra.metrics provides PrometheusInboxMetrics(prefix=None) and PrometheusOutboxMetrics(prefix=None) — implementations of InboxMetrics and OutboxMetrics. ProcessingMetrics is the shared base of those two protocols and has no implementation of its own. Wire them into the factories or pass to MetricsStep directly.

Version

from omni_box import __version__