Metadata-Version: 2.4
Name: outbox-pattern
Version: 0.1.1
Summary: A reliable, lightweight, framework-agnostic Transactional Outbox pattern implementation for Python (FastAPI, Flask, etc.).
Project-URL: Homepage, https://github.com/outbox-pattern/outbox-pattern
Project-URL: Repository, https://github.com/outbox-pattern/outbox-pattern
Author: Open Source Community
License: MIT
License-File: LICENSE
Keywords: aio-pika,beanie,event-driven,fastapi,flask,microservices,mongodb,outbox,outbox-pattern,outbox_pattern,pika,rabbitmq,transactional-outbox
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Framework :: Flask
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: aio-pika>=9.0.0
Requires-Dist: beanie>=1.20.0
Requires-Dist: pika>=1.3.0
Requires-Dist: pymongo>=4.0.0
Provides-Extra: all
Requires-Dist: logfire>=0.30.0; extra == 'all'
Provides-Extra: telemetry
Requires-Dist: logfire>=0.30.0; extra == 'telemetry'
Description-Content-Type: text/markdown

# Outbox Pattern SDK (`outbox-pattern`)

[![PyPI version](https://badge.fury.io/py/outbox-pattern.svg)](https://badge.fury.io/py/outbox-pattern)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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:

```bash
# 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:

```python
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:

```python
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:

```python
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**.
