Core (Outbox/Inbox)

nats_outbox.core.outbox

outbox_transaction — core context manager for the Transactional Outbox Pattern.

Design principles

1. Framework-agnostic: this module has zero FastAPI dependency. It only requires a SQLAlchemy AsyncSession. FastAPI integration is a thin layer on top (see examples/fastapi_integration.py).

2. Single transaction: both domain objects and outbox events are committed atomically in one SQL COMMIT. If the commit fails, neither is persisted — no partial state, no phantom events.

3. Nats-Msg-Id frozen at write time: the event_id (which becomes the Nats-Msg-Id NATS header) is generated when publish_event() is called, not when the relay publishes. This is the critical invariant:

   • Retry 1: relay publishes with Nats-Msg-Id = event_id → ACK received → OK
   • Retry 2 (after crash): same Nats-Msg-Id → JetStream dedup silently drops the duplicate within the dedup window. If we generated a new UUID per publish attempt, a lost ACK would produce a duplicate message in the stream — defeating the at-least-once guarantee.

4. Session lifecycle: outbox_transaction commits the session on aexit. The caller must NOT commit the session afterward. For FastAPI dependencies that normally handle commit/rollback, exclude the session from auto-commit when wrapping with outbox_transaction.

If you need to use outbox_transaction inside an existing async with session.begin(): block, call tx._flush_events() manually and let the outer transaction commit.

Trade-offs

• outbox_transaction always commits: simpler API, but means you can't stage multiple independent logical units then commit them all at once. For that use case, instantiate OutboxTransaction directly and call _flush_events() before your own commit.
• No nested transaction support in V1: using outbox_transaction inside an already-begun SQLAlchemy transaction will cause a double-commit. Documented as a known limitation.

OutboxTransaction

A thin wrapper around AsyncSession that adds outbox-aware event staging.

Call publish_event() to stage events. They are written to the database (within the same transaction as your domain objects) when the surrounding outbox_transaction context manager exits successfully.

The session object is still accessible via .session for advanced use cases (raw SQL, bulk inserts, etc.).

Source code in nats_outbox/core/outbox.py

class OutboxTransaction:
    """
    A thin wrapper around AsyncSession that adds outbox-aware event staging.

    Call publish_event() to stage events. They are written to the database
    (within the same transaction as your domain objects) when the surrounding
    outbox_transaction context manager exits successfully.

    The session object is still accessible via .session for advanced use cases
    (raw SQL, bulk inserts, etc.).
    """

    def __init__(self, session: AsyncSession) -> None:
        self._session = session
        self._pending_events: list[OutboxEvent] = []

    # ── Domain object proxy ──────────────────────────────────────────────────

    def add(self, instance: Any) -> None:
        """
        Add a domain object to the SQLAlchemy session.

        Mirrors session.add() — provided as a convenience so callers don't
        need to hold a reference to both tx and session separately.
        """
        self._session.add(instance)

    def add_all(self, instances: list[Any]) -> None:
        """Add multiple domain objects in one call."""
        for instance in instances:
            self._session.add(instance)

    # ── Event staging ────────────────────────────────────────────────────────

    def publish_event(
        self,
        subject: str,
        payload: dict[str, Any],
        *,
        aggregate_id: str | None = None,
        aggregate_type: str | None = None,
        scheduled_at: datetime | None = None,
        headers: dict[str, str] | None = None,
        event_id: uuid.UUID | None = None,
    ) -> uuid.UUID:
        """
        Stage an outbox event to be persisted within this transaction.

        Parameters
        ----------
        subject:
            NATS subject (e.g. "photo.created", "org.42.user.invited").
        payload:
            Event body. Must be JSON-serializable.
        aggregate_id:
            ID of the source entity (e.g. str(photo.id)). Used for ordering
            guarantees and observability. Optional but strongly recommended.
        aggregate_type:
            Class/type name of the source entity (e.g. "Photo"). Optional.
        scheduled_at:
            Earliest publish time. Defaults to now() for immediate delivery.
            Pass a future datetime for delayed publish.
        headers:
            Extra NATS headers to include. Nats-Msg-Id is always set to
            event_id and cannot be overridden here.
        event_id:
            Override the auto-generated UUID. Useful when you need the event_id
            before calling publish_event (e.g. to store it as a FK in a domain
            object). If omitted, a UUID4 is generated.

        Returns
        -------
        uuid.UUID
            The event_id that was assigned. Store this if you need to correlate
            the outbox row with consumer-side inbox deduplication.

        Nats-Msg-Id invariant
        ---------------------
        The Nats-Msg-Id header is set here, not in the publisher. The publisher
        reads it from OutboxEvent.headers and enforces it (no override). This
        guarantees the same ID is used across all retry attempts, enabling
        JetStream's dedup window to absorb retries transparently.
        """
        _event_id = event_id if event_id is not None else uuid.uuid4()

        _headers: dict[str, str] = dict(headers or {})

        # ── Critical invariant: Nats-Msg-Id = event_id, set at write time ──
        # Do NOT allow callers to override this via the headers param.
        # The publisher also enforces this as a defense-in-depth safeguard.
        _headers["Nats-Msg-Id"] = str(_event_id)

        event = OutboxEvent(
            event_id=_event_id,
            subject=subject,
            payload=payload,
            headers=_headers,
            aggregate_id=aggregate_id,
            aggregate_type=aggregate_type,
            scheduled_at=scheduled_at or datetime.now(tz=UTC),
            status="pending",
            retry_count=0,
        )
        self._pending_events.append(event)
        return _event_id

    # ── Internal ─────────────────────────────────────────────────────────────

    async def _flush_events(self) -> None:
        """
        Add all staged OutboxEvent instances to the session.

        Called by the context manager before commit. Can be called manually
        if you manage the transaction yourself (e.g. inside an outer begin()).
        """
        for event in self._pending_events:
            self._session.add(event)
        # Keep the list in case of a re-flush after a savepoint rollback.
        # The session deduplicates by identity map.

    @property
    def session(self) -> AsyncSession:
        """
        Escape hatch: the underlying SQLAlchemy session.

        Use for operations not covered by the OutboxTransaction proxy
        (e.g. raw SQL, bulk operations, SQLAlchemy-specific features).
        """
        return self._session

    @property
    def staged_events(self) -> list[OutboxEvent]:
        """Read-only view of staged (not yet committed) events. Useful for testing."""
        return list(self._pending_events)

session property

Escape hatch: the underlying SQLAlchemy session.

Use for operations not covered by the OutboxTransaction proxy (e.g. raw SQL, bulk operations, SQLAlchemy-specific features).

staged_events property

Read-only view of staged (not yet committed) events. Useful for testing.

add(instance)

Add a domain object to the SQLAlchemy session.

Mirrors session.add() — provided as a convenience so callers don't need to hold a reference to both tx and session separately.

Source code in nats_outbox/core/outbox.py

def add(self, instance: Any) -> None:
    """
    Add a domain object to the SQLAlchemy session.

    Mirrors session.add() — provided as a convenience so callers don't
    need to hold a reference to both tx and session separately.
    """
    self._session.add(instance)

add_all(instances)

Add multiple domain objects in one call.

Source code in nats_outbox/core/outbox.py

def add_all(self, instances: list[Any]) -> None:
    """Add multiple domain objects in one call."""
    for instance in instances:
        self._session.add(instance)

publish_event(subject, payload, *, aggregate_id=None, aggregate_type=None, scheduled_at=None, headers=None, event_id=None)

Stage an outbox event to be persisted within this transaction.

Parameters:

Name	Type	Description	Default
subject	str	NATS subject (e.g. "photo.created", "org.42.user.invited").	required
payload	dict[str, Any]	Event body. Must be JSON-serializable.	required
aggregate_id	str | None	ID of the source entity (e.g. str(photo.id)). Used for ordering guarantees and observability. Optional but strongly recommended.	None
aggregate_type	str | None	Class/type name of the source entity (e.g. "Photo"). Optional.	None
scheduled_at	datetime | None	Earliest publish time. Defaults to now() for immediate delivery. Pass a future datetime for delayed publish.	None
headers	dict[str, str] | None	Extra NATS headers to include. Nats-Msg-Id is always set to event_id and cannot be overridden here.	None
event_id	UUID | None	Override the auto-generated UUID. Useful when you need the event_id before calling publish_event (e.g. to store it as a FK in a domain object). If omitted, a UUID4 is generated.	None

Returns:

Type	Description
UUID	The event_id that was assigned. Store this if you need to correlate the outbox row with consumer-side inbox deduplication.

Nats-Msg-Id invariant

The Nats-Msg-Id header is set here, not in the publisher. The publisher reads it from OutboxEvent.headers and enforces it (no override). This guarantees the same ID is used across all retry attempts, enabling JetStream's dedup window to absorb retries transparently.

Source code in nats_outbox/core/outbox.py

def publish_event(
    self,
    subject: str,
    payload: dict[str, Any],
    *,
    aggregate_id: str | None = None,
    aggregate_type: str | None = None,
    scheduled_at: datetime | None = None,
    headers: dict[str, str] | None = None,
    event_id: uuid.UUID | None = None,
) -> uuid.UUID:
    """
    Stage an outbox event to be persisted within this transaction.

    Parameters
    ----------
    subject:
        NATS subject (e.g. "photo.created", "org.42.user.invited").
    payload:
        Event body. Must be JSON-serializable.
    aggregate_id:
        ID of the source entity (e.g. str(photo.id)). Used for ordering
        guarantees and observability. Optional but strongly recommended.
    aggregate_type:
        Class/type name of the source entity (e.g. "Photo"). Optional.
    scheduled_at:
        Earliest publish time. Defaults to now() for immediate delivery.
        Pass a future datetime for delayed publish.
    headers:
        Extra NATS headers to include. Nats-Msg-Id is always set to
        event_id and cannot be overridden here.
    event_id:
        Override the auto-generated UUID. Useful when you need the event_id
        before calling publish_event (e.g. to store it as a FK in a domain
        object). If omitted, a UUID4 is generated.

    Returns
    -------
    uuid.UUID
        The event_id that was assigned. Store this if you need to correlate
        the outbox row with consumer-side inbox deduplication.

    Nats-Msg-Id invariant
    ---------------------
    The Nats-Msg-Id header is set here, not in the publisher. The publisher
    reads it from OutboxEvent.headers and enforces it (no override). This
    guarantees the same ID is used across all retry attempts, enabling
    JetStream's dedup window to absorb retries transparently.
    """
    _event_id = event_id if event_id is not None else uuid.uuid4()

    _headers: dict[str, str] = dict(headers or {})

    # ── Critical invariant: Nats-Msg-Id = event_id, set at write time ──
    # Do NOT allow callers to override this via the headers param.
    # The publisher also enforces this as a defense-in-depth safeguard.
    _headers["Nats-Msg-Id"] = str(_event_id)

    event = OutboxEvent(
        event_id=_event_id,
        subject=subject,
        payload=payload,
        headers=_headers,
        aggregate_id=aggregate_id,
        aggregate_type=aggregate_type,
        scheduled_at=scheduled_at or datetime.now(tz=UTC),
        status="pending",
        retry_count=0,
    )
    self._pending_events.append(event)
    return _event_id

outbox_transaction(session) async

Async context manager that wraps a SQLAlchemy session with outbox support.

On successful exit (aexit with no exception): 1. All staged OutboxEvent rows are added to the session. 2. session.commit() is called — domain objects AND events are persisted atomically.

On exception: session.rollback() is called — neither domain objects nor events are persisted.

Usage (framework-agnostic): ::

engine = create_async_engine(settings.database_url)
async_session = async_sessionmaker(engine)

async with async_session() as session:
    async with outbox_transaction(session) as tx:
        photo = Photo(url="https://...", user_id=42)
        tx.add(photo)
        tx.publish_event(
            subject="photo.created",
            payload={"photo_id": str(photo.id)},
            aggregate_id=str(photo.id),
            aggregate_type="Photo",
        )
    # session is committed (and closed by the outer async with)

FastAPI dependency injection: ::

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSession(engine) as session:
        yield session
        # NOTE: do NOT commit here — outbox_transaction does it.

@router.post("/photos")
async def create_photo(session: AsyncSession = Depends(get_session)):
    async with outbox_transaction(session) as tx:
        photo = Photo(url=request.url)
        tx.add(photo)
        await session.flush()  # flush to get photo.id if needed
        tx.publish_event(
            subject="photo.created",
            payload={"photo_id": str(photo.id)},
        )

Known limitation (V1): Do NOT use inside an existing async with session.begin(): block. outbox_transaction will issue a commit that also commits the outer transaction. If you need composable transactions, call await tx._flush_events() and manage the commit yourself.

Source code in nats_outbox/core/outbox.py

@asynccontextmanager
async def outbox_transaction(
    session: AsyncSession,
) -> AsyncGenerator[OutboxTransaction, None]:
    """
    Async context manager that wraps a SQLAlchemy session with outbox support.

    On successful exit (__aexit__ with no exception):
        1. All staged OutboxEvent rows are added to the session.
        2. session.commit() is called — domain objects AND events are persisted atomically.

    On exception:
        session.rollback() is called — neither domain objects nor events are persisted.

    Usage (framework-agnostic):
    ::

        engine = create_async_engine(settings.database_url)
        async_session = async_sessionmaker(engine)

        async with async_session() as session:
            async with outbox_transaction(session) as tx:
                photo = Photo(url="https://...", user_id=42)
                tx.add(photo)
                tx.publish_event(
                    subject="photo.created",
                    payload={"photo_id": str(photo.id)},
                    aggregate_id=str(photo.id),
                    aggregate_type="Photo",
                )
            # session is committed (and closed by the outer async with)

    FastAPI dependency injection:
    ::

        async def get_session() -> AsyncGenerator[AsyncSession, None]:
            async with AsyncSession(engine) as session:
                yield session
                # NOTE: do NOT commit here — outbox_transaction does it.

        @router.post("/photos")
        async def create_photo(session: AsyncSession = Depends(get_session)):
            async with outbox_transaction(session) as tx:
                photo = Photo(url=request.url)
                tx.add(photo)
                await session.flush()  # flush to get photo.id if needed
                tx.publish_event(
                    subject="photo.created",
                    payload={"photo_id": str(photo.id)},
                )

    Known limitation (V1):
        Do NOT use inside an existing `async with session.begin():` block.
        outbox_transaction will issue a commit that also commits the outer
        transaction. If you need composable transactions, call
        `await tx._flush_events()` and manage the commit yourself.
    """
    tx = OutboxTransaction(session)
    try:
        yield tx
        await tx._flush_events()
        await session.commit()
    except Exception:
        await session.rollback()
        raise

nats_outbox.core.inbox

Inbox Pattern — consumer-side event deduplication.

Why this is needed

The outbox relay provides at-least-once delivery: an event will eventually be published, but may be published more than once (e.g. relay crash after publish but before marking as published). JetStream's Nats-Msg-Id dedup window handles duplicates within the dedup window (default 2 minutes), but consumers must handle duplicates that arrive outside that window or via different streams.

Strategy

INSERT INTO inbox_events (event_id, consumer_group) ON CONFLICT DO NOTHING

This is an atomic check-and-insert: if the row already exists, the INSERT returns zero rows (conflict), indicating the event is a duplicate. No SELECT is needed — avoids the TOCTOU race condition of SELECT then INSERT.

The consumer_group column lets multiple consumers (e.g. "billing", "analytics") independently deduplicate events with the same event_id — each group tracks its own processed events.

Usage

::

async with AsyncSession(engine) as session:
    async with session.begin():
        deduplicator = InboxDeduplicator(session, consumer_group="billing")
        if await deduplicator.is_duplicate(event_id):
            return  # already processed, skip

        # ... process event business logic here ...
        # The inbox row + business effect commit together atomically.

Table schema (run create_inbox_table() or use the provided migration) ::

CREATE TABLE inbox_events (
    id              BIGSERIAL    PRIMARY KEY,
    event_id        UUID         NOT NULL,
    consumer_group  TEXT         NOT NULL,
    processed_at    TIMESTAMPTZ  NOT NULL DEFAULT now(),
    CONSTRAINT uq_inbox_event UNIQUE (event_id, consumer_group)
);

InboxDeduplicator

Stateless helper for consumer-side event deduplication.

Thread-safety: one instance per request/task — do not share across concurrent coroutines with the same session.

Source code in nats_outbox/core/inbox.py

class InboxDeduplicator:
    """
    Stateless helper for consumer-side event deduplication.

    Thread-safety: one instance per request/task — do not share across
    concurrent coroutines with the same session.
    """

    def __init__(self, session: AsyncSession, consumer_group: str) -> None:
        self._session = session
        self._consumer_group = consumer_group

    async def is_duplicate(self, event_id: uuid.UUID) -> bool:
        """
        Check whether event_id has already been processed by this consumer group.

        Uses INSERT ... ON CONFLICT DO NOTHING — atomic, no SELECT needed.
        Returns True if the event is a duplicate (already in inbox_events).
        Returns False and inserts the row if the event is new.

        IMPORTANT: This must be called within an open transaction. The inbox row
        and the business effect must commit together, otherwise a crash between
        the two commits can leave the inbox row without the business effect
        (or vice versa).
        """
        result = await self._session.execute(
            text("""
                INSERT INTO inbox_events (event_id, consumer_group)
                VALUES (:event_id, :consumer_group)
                ON CONFLICT (event_id, consumer_group) DO NOTHING
                RETURNING id
            """),
            {
                "event_id": str(event_id),
                "consumer_group": self._consumer_group,
            },
        )
        row = result.fetchone()
        # row is None ↔ conflict (already processed) → duplicate
        # row is not None ↔ inserted successfully → new event
        return row is None

is_duplicate(event_id) async

Check whether event_id has already been processed by this consumer group.

Uses INSERT ... ON CONFLICT DO NOTHING — atomic, no SELECT needed. Returns True if the event is a duplicate (already in inbox_events). Returns False and inserts the row if the event is new.

IMPORTANT: This must be called within an open transaction. The inbox row and the business effect must commit together, otherwise a crash between the two commits can leave the inbox row without the business effect (or vice versa).

Source code in nats_outbox/core/inbox.py

async def is_duplicate(self, event_id: uuid.UUID) -> bool:
    """
    Check whether event_id has already been processed by this consumer group.

    Uses INSERT ... ON CONFLICT DO NOTHING — atomic, no SELECT needed.
    Returns True if the event is a duplicate (already in inbox_events).
    Returns False and inserts the row if the event is new.

    IMPORTANT: This must be called within an open transaction. The inbox row
    and the business effect must commit together, otherwise a crash between
    the two commits can leave the inbox row without the business effect
    (or vice versa).
    """
    result = await self._session.execute(
        text("""
            INSERT INTO inbox_events (event_id, consumer_group)
            VALUES (:event_id, :consumer_group)
            ON CONFLICT (event_id, consumer_group) DO NOTHING
            RETURNING id
        """),
        {
            "event_id": str(event_id),
            "consumer_group": self._consumer_group,
        },
    )
    row = result.fetchone()
    # row is None ↔ conflict (already processed) → duplicate
    # row is not None ↔ inserted successfully → new event
    return row is None

InboxEvent

Bases: Base

Records events that have been successfully processed by a consumer group. Enables idempotent consumers without application-level locking.

Source code in nats_outbox/core/inbox.py

class InboxEvent(Base):
    """
    Records events that have been successfully processed by a consumer group.
    Enables idempotent consumers without application-level locking.
    """

    __tablename__ = "inbox_events"
    __table_args__ = (
        UniqueConstraint(
            "event_id",
            "consumer_group",
            name="uq_inbox_event_consumer",
        ),
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    event_id: Mapped[uuid.UUID] = mapped_column(
        PG_UUID(as_uuid=True),
        nullable=False,
        comment="Matches OutboxEvent.event_id from the publisher.",
    )
    consumer_group: Mapped[str] = mapped_column(
        Text,
        nullable=False,
        comment=(
            "Logical consumer name (e.g. 'billing-service', 'notification-worker'). "
            "Allows multiple consumers to independently deduplicate the same event."
        ),
    )
    processed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
    )

    def __repr__(self) -> str:
        return f"<InboxEvent event_id={self.event_id} consumer_group={self.consumer_group!r}>"

create_inbox_table(engine) async

Create the inbox_events table. For tests and quick-start scripts.

Source code in nats_outbox/core/inbox.py

async def create_inbox_table(engine: Any) -> None:
    """Create the inbox_events table. For tests and quick-start scripts."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

nats_outbox.core.models

SQLAlchemy ORM model for the outbox_events table.

Schema design decisions (full rationale in the project README):

• BIGSERIAL PK: sequential → zero B-tree fragmentation at high insert rates. UUID is present as a business key (event_id), not the structural PK.

• JSONB (not JSON): stored as binary → faster reads, supports GIN indexes for observability queries (e.g. payload @> '{"user_id": 42}').

• status TEXT + CHECK (not ENUM): adding a new status value with ENUM requires ALTER TYPE which takes AccessExclusiveLock. TEXT + CHECK supports ALTER TABLE ... ADD CONSTRAINT NOT VALID + VALIDATE in two lock-free steps.

• updated_at: maintained via SQLAlchemy's onupdate mechanism (client-side, evaluated on every ORM UPDATE). Useful for debugging stuck events without parsing last_error timestamps.

• scheduled_at separate from created_at: enables delayed/scheduled publish at zero extra implementation cost in the relay (just check scheduled_at <= now()).

• Three partial indexes (pending, published, aggregate) keep index size small because the majority of rows are in status='published'.

Base

Bases: DeclarativeBase

Shared declarative base. Import this into your own Base if you want outbox_events co-located with your application tables.

Source code in nats_outbox/core/models.py

class Base(DeclarativeBase):
    """Shared declarative base. Import this into your own Base if you want
    outbox_events co-located with your application tables."""

OutboxEvent

Bases: Base

Represents a single event staged in the transactional outbox.

State machine: pending ──(publish ok)──► published pending ──(retry exhausted)──► failed

The relay only reads rows in status='pending'. Published rows are kept for retention_days then purged by the cleanup task. Failed rows are kept indefinitely (alert on them, then resolve manually).

Source code in nats_outbox/core/models.py

class OutboxEvent(Base):
    """
    Represents a single event staged in the transactional outbox.

    State machine:
        pending ──(publish ok)──► published
        pending ──(retry exhausted)──► failed

    The relay only reads rows in status='pending'.
    Published rows are kept for `retention_days` then purged by the cleanup task.
    Failed rows are kept indefinitely (alert on them, then resolve manually).
    """

    __tablename__ = "outbox_events"
    __table_args__ = (
        CheckConstraint(
            "status IN ('pending', 'published', 'failed')",
            name="chk_outbox_status",
        ),
        UniqueConstraint("event_id", name="uq_outbox_event_id"),
        # ── Partial index: polling relay query ──────────────────────────────
        # The relay's hot path is:
        #   WHERE status = 'pending' AND scheduled_at <= now()
        #   ORDER BY scheduled_at, id
        #   LIMIT batch_size
        #   FOR UPDATE SKIP LOCKED
        #
        # The partial index covers status='pending' only → much smaller than
        # a full-table index once most rows are published.
        # Postgres applies the `scheduled_at <= now()` filter as a post-scan
        # predicate on the (already-tiny) partial index — verified acceptable
        # via EXPLAIN ANALYZE at production volumes.
        Index(
            "idx_outbox_pending",
            "scheduled_at",
            "id",
            postgresql_where=text("status = 'pending'"),
        ),
        # ── Partial index: cleanup / retention ──────────────────────────────
        # DELETE WHERE status = 'published' AND published_at < cutoff
        Index(
            "idx_outbox_cleanup",
            "published_at",
            postgresql_where=text("status = 'published'"),
        ),
        # ── Partial index: aggregate ordering ───────────────────────────────
        # Retrieve pending events for a specific aggregate in insertion order.
        # Used by observability queries and WAL tailing (V2).
        Index(
            "idx_outbox_aggregate",
            "aggregate_type",
            "aggregate_id",
            "id",
            postgresql_where=text("status = 'pending'"),
        ),
    )

    # ── Identity ─────────────────────────────────────────────────────────────
    id: Mapped[int] = mapped_column(
        BigInteger,
        primary_key=True,
        autoincrement=True,
        comment="Internal sequential PK. Used for ordering and FOR UPDATE SKIP LOCKED.",
    )
    event_id: Mapped[uuid.UUID] = mapped_column(
        PG_UUID(as_uuid=True),
        nullable=False,
        default=uuid.uuid4,
        comment=(
            "Business identifier. Exposed to consumers for idempotence (Inbox Pattern). "
            "Also used verbatim as the NATS Nats-Msg-Id header — stable across retries "
            "so JetStream dedup prevents duplicates even on ACK loss."
        ),
    )

    # ── NATS routing ─────────────────────────────────────────────────────────
    subject: Mapped[str] = mapped_column(
        Text,
        nullable=False,
        comment="NATS subject (e.g. 'photo.created', 'org.{id}.user.invited').",
    )
    headers: Mapped[dict[str, Any]] = mapped_column(
        JSONB,
        nullable=False,
        default=dict,
        comment=(
            "NATS message headers. Always includes Nats-Msg-Id=event_id. "
            "Extend with tracing headers (traceparent, X-Correlation-Id) as needed."
        ),
    )

    # ── Content ──────────────────────────────────────────────────────────────
    payload: Mapped[dict[str, Any]] = mapped_column(
        JSONB,
        nullable=False,
        comment="Event payload. JSONB for binary storage and GIN-indexable queries.",
    )

    # ── Lifecycle ────────────────────────────────────────────────────────────
    status: Mapped[str] = mapped_column(
        String(20),
        nullable=False,
        default="pending",
        comment="State machine: pending → published | failed.",
    )
    retry_count: Mapped[int] = mapped_column(
        SmallInteger,
        nullable=False,
        default=0,
        comment=(
            "Number of failed publish attempts so far. "
            "V1 trade-off: max_retries is a global setting (OutboxSettings), "
            "not per-event. Per-event max_retries deferred to V2."
        ),
    )
    last_error: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment="Truncated repr of the last exception. Aids debugging without log-diving.",
    )

    # ── Correlation & ordering ────────────────────────────────────────────────
    aggregate_id: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment=(
            "ID of the source aggregate (e.g. photo_id, user_id). TEXT covers UUIDs, slugs, ints."
        ),
    )
    aggregate_type: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment="Type name of the source aggregate (e.g. 'Photo', 'User').",
    )

    # ── Timestamps ───────────────────────────────────────────────────────────
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
        comment="When the event was written to the outbox (within the business transaction).",
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
        comment=(
            "Last time this row was modified by the relay (set on every ORM UPDATE). "
            "Useful for detecting stuck events: a row that stays pending for a long time "
            "with a recent updated_at is actively being retried; an old updated_at means "
            "the relay is not reaching this event."
        ),
    )
    scheduled_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
        comment=(
            "Earliest timestamp at which the relay should attempt to publish. "
            "Default = now() (immediate). Set to a future time for delayed publish. "
            "The relay pushes this forward on retry using exponential backoff."
        ),
    )
    published_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
        comment=(
            "When the event was successfully ACK'd by JetStream. "
            "NULL until published. Used to compute publish latency metrics "
            "and to identify rows eligible for retention cleanup."
        ),
    )

    def __repr__(self) -> str:
        return (
            f"<OutboxEvent id={self.id} event_id={self.event_id} "
            f"subject={self.subject!r} status={self.status} retries={self.retry_count}>"
        )

create_tables(engine) async

Create all tables defined in this module and in core.inbox.

Imports InboxEvent here to ensure its table is registered with Base.metadata before create_all() is called.

Convenience helper for tests and quick-start scripts. For production, prefer Alembic migrations generated from this model.

Source code in nats_outbox/core/models.py

async def create_tables(engine: Any) -> None:
    """
    Create all tables defined in this module and in core.inbox.

    Imports InboxEvent here to ensure its table is registered with
    Base.metadata before create_all() is called.

    Convenience helper for tests and quick-start scripts.
    For production, prefer Alembic migrations generated from this model.
    """
    # Import InboxEvent so it registers with Base.metadata
    from nats_outbox.core.inbox import InboxEvent  # noqa: F401

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

drop_tables(engine) async

Drop all tables. For tests only — never call in production.

Source code in nats_outbox/core/models.py

async def drop_tables(engine: Any) -> None:
    """Drop all tables. For tests only — never call in production."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)