outbox-pattern 0.1.2

A reliable, lightweight, framework-agnostic Transactional Outbox pattern implementation for Python (FastAPI, Flask, etc.).

Outbox Pattern SDK (outbox-pattern)

A lightweight, reliable, framework-agnostic implementation of the Transactional Outbox Pattern in Python. Built to work seamlessly with FastAPI (Async), Flask (Sync), or any custom Python application.

---

Key Features

• 🔄 Transactional Outbox Pattern: Ensures event publication consistency by storing outbox events atomically before dispatching.
• 🚀 Framework Agnostic: Native support for FastAPI (asyncio lifespan) and Flask (threading daemon).
• ⚡ Async & Sync Support: Native implementations for both asynchronous (asyncio) and synchronous (threading) workloads.
• 🐇 RabbitMQ Integration: Built-in publishers for aio-pika (async) and pika (sync).
• 🍃 MongoDB Integration: Outbox repositories powered by beanie (async) and pymongo (sync).
• 📊 Telemetry & Observability: Integrated provider for Pydantic Logfire with structured logs and dead-letter event tracing.
• 🛡️ Distributed Lock Safety: Prevents race conditions among multiple concurrent worker instances using optimistic lock timeouts.

---

Installation

Install the base package or specify optional extras depending on your stack:

# Install for FastAPI (Async: Beanie + aio-pika)
pip install "outbox-pattern[async]"

# Install for Flask (Sync: PyMongo + pika)
pip install "outbox-pattern[sync]"

# Install with Telemetry (Logfire)
pip install "outbox-pattern[telemetry]"

# Install all optional dependencies
pip install "outbox-pattern[all]"

---

Framework Integration Examples

1. FastAPI Integration (Async)

FastAPI applications manage background workers using the lifespan context manager:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel

from outbox_pattern import OutboxEvent
from outbox_pattern.async_impl import (
    AsyncBeanieRepository,
    AsyncAioPikaPublisher,
    AsyncOutboxWorker,
)

# Repositories & Workers
repository = AsyncBeanieRepository()
publisher = AsyncAioPikaPublisher(broker_url="amqp://guest:guest@localhost:5672/")
worker = AsyncOutboxWorker(repository=repository, publisher=publisher, poll_interval=3)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Start the Outbox worker task
    await worker.start()
    yield
    # Shutdown: Stop worker gracefully
    await worker.stop()

app = FastAPI(title="My FastAPI Microservice", lifespan=lifespan)

class CreateUserDTO(BaseModel):
    user_id: str
    email: str

@app.post("/users")
async def create_user(dto: CreateUserDTO):
    # 1. Save user inside main DB transaction...
    
    # 2. Save Outbox Event atomically
    event = OutboxEvent(
        tenant="tenant_default",
        topic="user.registered",
        payload=dto.model_dump(),
    )
    await repository.save_event(event)
    
    return {"status": "user_created", "user_id": dto.user_id}

---

2. Flask Integration (Sync)

Flask applications use daemon background threads for outbox polling:

from flask import Flask, request, jsonify
from pymongo import MongoClient

from outbox_pattern import OutboxEvent
from outbox_pattern.sync_impl import (
    SyncPyMongoRepository,
    SyncPikaPublisher,
    SyncOutboxWorker,
)

app = Flask(__name__)

# Initialize MongoDB & RabbitMQ
mongo_client = MongoClient("mongodb://localhost:27017/")
db = mongo_client["my_flask_app"]

repository = SyncPyMongoRepository(collection=db["outbox_events"])
publisher = SyncPikaPublisher(broker_url="amqp://guest:guest@localhost:5672/")

# Start Outbox worker in a background daemon thread
worker = SyncOutboxWorker(repository=repository, publisher=publisher, poll_interval=5)
worker.start()

@app.route("/orders", methods=["POST"])
def create_order():
    data = request.json
    
    # 1. Save order in MongoDB...
    
    # 2. Record Outbox Event
    event = OutboxEvent(
        tenant=data.get("tenant", "default"),
        topic="order.created",
        payload=data,
    )
    repository.save_event(event)
    
    return jsonify({"message": "Order created successfully"}), 201

---

Observability & Telemetry

Enable Pydantic Logfire telemetry to trace event publishing lifecycle and dead-letter queues:

from outbox_pattern.telemetry import LogfireTelemetryProvider

telemetry = LogfireTelemetryProvider()

worker = AsyncOutboxWorker(
    repository=repository,
    publisher=publisher,
    telemetry=telemetry,
)

---

Configuration

Environment variables can be set to override default behavior:

Environment Variable	Default	Description
OUTBOX_MAX_RETRIES	3	Maximum processing attempts before marking an event as failed.
OUTBOX_BATCH_SIZE	100	Number of events acquired per polling loop iteration.
OUTBOX_LOCK_TIMEOUT_SECONDS	300	Lock expiration time (in seconds) for stale workers.

---

License

This project is licensed under the terms of the MIT License.