Implement the Transactional Outbox Pattern with SQLite in Python

import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json

@dataclass
class Order:
    order_id: str
    amount: float
    status: str

class TransactionalOutbox:
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self._create_tables()

    def _create_tables(self):
        with self.conn:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS orders (
                    id TEXT PRIMARY KEY,
                    amount REAL,
                    status TEXT
                )
            """)
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS outbox (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    payload TEXT,
                    created_at TEXT,
                    processed INTEGER DEFAULT 0
                )
            """)

    def create_order(self, order: Order):
        with self.conn:
            # Insert order in the same transaction as outbox event
            self.conn.execute(
                "INSERT INTO orders VALUES (?, ?, ?)",
                (order.order_id, order.amount, order.status)
            )
            event = {
                "type": "order_created",
                "order_id": order.order_id,
                "amount": order.amount
            }
            self.conn.execute(
                "INSERT INTO outbox (payload, created_at) VALUES (?, ?)",
                (json.dumps(event), datetime.now(timezone.utc).isoformat())
            )

    def poll_outbox(self):
        with self.conn:
            events = self.conn.execute(
                "SELECT id, payload FROM outbox WHERE processed = 0 LIMIT 10"
            ).fetchall()
            for event_id, payload in events:
                print(f"Publishing: {payload}")
                self.conn.execute(
                    "UPDATE outbox SET processed = 1 WHERE id = ?",
                    (event_id,)
                )

if __name__ == "__main__":
    outbox = TransactionalOutbox()
    outbox.create_order(Order("ORD-1", 99.50, "pending"))
    outbox.create_order(Order("ORD-2", 250.00, "pending"))
    
    print("Outbox contents before processing:")
    events = outbox.conn.execute("SELECT payload FROM outbox").fetchall()
    for row in events:
        print(row[0])
    
    print("\nPolling outbox:")
    outbox.poll_outbox()