Skip to main content

How to Use the Transactional Outbox

The Transactional Outbox Pattern ensures reliable coordination between SQL database operations and external systems (such as message buses, external search engines, or third-party APIs) without risking data inconsistency or requiring heavy distributed transactions.

This guide explains how to queue entries in the outbox and how the background process dispatches them.

Staging Entries in the Outbox​

When performing database updates inside a unit of work, write to the outbox repository within the same database transaction.

from framework_m_core.domain.outbox import OutboxEntry
from framework_m_standard.adapters.db.outbox_repository import OutboxRepository

async with UnitOfWork(session_factory) as uow:
    # 1. Update your primary entity
    await repo.save(uow.session, entity)

    # 2. Stage event payload in the outbox
    outbox_entry = OutboxEntry(
        target="customer.created",
        payload={"customer_id": str(entity.id), "email": entity.email}
    )
    
    # Write to outbox using the same transaction session
    await outbox_repo.add(uow.session, outbox_entry)
    
    await uow.commit() # Committed atomically

How Outbox Entries are Processed​

A background outbox worker (process_outbox) periodically retrieves pending entries, publishes them to the configured EventBus, and updates the status:

  • Pending: Waiting to be picked up by the worker.
  • Processed: Dispatched to the target destination successfully.
  • Failed: Encountered an error during processing. The worker records error_message and increments retry_count.

Running the Outbox Worker​

The outbox processor is a task-scheduled job. You can start the worker pool to process outbox entries:

uv run m worker --pool default

Or trigger the outbox processing job programmatically:

from framework_m_standard.adapters.jobs.outbox_worker import process_outbox

# Process a batch of up to 100 pending entries
await process_outbox(batch_size=100)