Distributed Data Patterns

Introduction

When each service owns its database, you lose ACID transactions across service boundaries. From working on distributed systems, I've found that traditional two-phase commit doesn't scale. Instead, we need patterns like Sagas for distributed transactions, Event Sourcing for audit trails, and CQRS for read/write optimization.

This article covers practical implementations of these patterns with Python.

The Distributed Transaction Problem

Why 2PC Doesn't Work

Issue	Description
Blocking	All participants lock resources until coordinator decides
Single point of failure	Coordinator failure blocks all transactions
Network partitions	Participants may be unable to reach coordinator
Latency	Multiple round trips increase response time
Availability	All services must be available

Saga Pattern

Choreography-Based Saga

from dataclasses import dataclass
from enum import Enum
from typing import Callable
import asyncio


class SagaState(Enum):
    STARTED = "started"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"


@dataclass
class SagaStep:
    name: str
    execute: Callable
    compensate: Callable


class ChoreographySaga:
    """Choreography-based saga using events."""
    
    def __init__(self, event_publisher, event_consumer):
        self.event_publisher = event_publisher
        self.event_consumer = event_consumer
        self.handlers = {}
    
    def on_event(self, event_type: str):
        """Register event handler."""
        def decorator(handler):
            self.handlers[event_type] = handler
            return handler
        return decorator
    
    async def start(self):
        """Start listening for events."""
        for event_type in self.handlers:
            await self.event_consumer.subscribe(
                event_type,
                self.handlers[event_type],
            )


# Order Service - initiates saga
class OrderSaga(ChoreographySaga):
    def __init__(self, order_repo, event_publisher, event_consumer):
        super().__init__(event_publisher, event_consumer)
        self.order_repo = order_repo
        self.setup_handlers()
    
    def setup_handlers(self):
        @self.on_event("order.creation_requested")
        async def handle_creation_request(event: dict):
            order = await self.create_order(event["data"])
            await self.event_publisher.publish(
                "order.created",
                {"order_id": order.id, **order.__dict__},
            )
        
        @self.on_event("payment.processed")
        async def handle_payment_success(event: dict):
            order_id = event["data"]["order_id"]
            await self.order_repo.update_status(order_id, "payment_confirmed")
            # Continue saga - request inventory
            await self.event_publisher.publish(
                "inventory.reservation_requested",
                {"order_id": order_id, "items": event["data"]["items"]},
            )
        
        @self.on_event("payment.failed")
        async def handle_payment_failure(event: dict):
            order_id = event["data"]["order_id"]
            await self.order_repo.update_status(order_id, "cancelled")
            await self.event_publisher.publish(
                "order.cancelled",
                {"order_id": order_id, "reason": "payment_failed"},
            )
        
        @self.on_event("inventory.reserved")
        async def handle_inventory_success(event: dict):
            order_id = event["data"]["order_id"]
            await self.order_repo.update_status(order_id, "confirmed")
            await self.event_publisher.publish(
                "order.confirmed",
                {"order_id": order_id},
            )
        
        @self.on_event("inventory.reservation_failed")
        async def handle_inventory_failure(event: dict):
            order_id = event["data"]["order_id"]
            # Compensate - request payment refund
            await self.event_publisher.publish(
                "payment.refund_requested",
                {"order_id": order_id, "reason": "inventory_unavailable"},
            )
            await self.order_repo.update_status(order_id, "cancelled")


# Payment Service - handles payment events
class PaymentSaga(ChoreographySaga):
    def setup_handlers(self):
        @self.on_event("order.created")
        async def handle_order_created(event: dict):
            order_data = event["data"]
            try:
                payment = await self.process_payment(
                    order_id=order_data["order_id"],
                    amount=order_data["total"],
                    customer_id=order_data["customer_id"],
                )
                await self.event_publisher.publish(
                    "payment.processed",
                    {"order_id": order_data["order_id"], "payment_id": payment.id},
                )
            except PaymentError as e:
                await self.event_publisher.publish(
                    "payment.failed",
                    {"order_id": order_data["order_id"], "error": str(e)},
                )
        
        @self.on_event("payment.refund_requested")
        async def handle_refund_request(event: dict):
            order_id = event["data"]["order_id"]
            await self.refund_payment(order_id)
            await self.event_publisher.publish(
                "payment.refunded",
                {"order_id": order_id},
            )

Orchestration-Based Saga

from dataclasses import dataclass, field
from typing import Any


@dataclass
class SagaStepResult:
    success: bool
    data: Any = None
    error: str | None = None


class SagaOrchestrator:
    """Centralized saga orchestrator."""
    
    def __init__(self, saga_id: str):
        self.saga_id = saga_id
        self.steps: list[SagaStep] = []
        self.completed_steps: list[str] = []
        self.state = SagaState.STARTED
    
    def add_step(
        self,
        name: str,
        execute: Callable,
        compensate: Callable,
    ):
        self.steps.append(SagaStep(name, execute, compensate))
    
    async def execute(self, context: dict) -> dict:
        """Execute saga steps in sequence."""
        results = {}
        
        for step in self.steps:
            try:
                result = await step.execute(context)
                
                if not result.success:
                    # Step failed, start compensation
                    await self._compensate(context, step.name)
                    self.state = SagaState.FAILED
                    return {"success": False, "failed_at": step.name, "error": result.error}
                
                self.completed_steps.append(step.name)
                results[step.name] = result.data
                context.update(result.data or {})
            
            except Exception as e:
                await self._compensate(context, step.name)
                self.state = SagaState.FAILED
                return {"success": False, "failed_at": step.name, "error": str(e)}
        
        self.state = SagaState.COMPLETED
        return {"success": True, "results": results}
    
    async def _compensate(self, context: dict, failed_step: str):
        """Execute compensation in reverse order."""
        self.state = SagaState.COMPENSATING
        
        # Get steps to compensate (completed ones in reverse)
        steps_to_compensate = [
            s for s in reversed(self.steps)
            if s.name in self.completed_steps
        ]
        
        for step in steps_to_compensate:
            try:
                await step.compensate(context)
            except Exception as e:
                # Log compensation failure - may need manual intervention
                logger.error(f"Compensation failed for {step.name}: {e}")


# Usage: Order Creation Saga
class OrderSagaOrchestrator:
    def __init__(
        self,
        order_service,
        payment_service,
        inventory_service,
        notification_service,
    ):
        self.order_service = order_service
        self.payment_service = payment_service
        self.inventory_service = inventory_service
        self.notification_service = notification_service
    
    async def create_order(self, order_data: dict) -> dict:
        saga = SagaOrchestrator(f"order-{uuid.uuid4()}")
        
        # Step 1: Create Order
        saga.add_step(
            "create_order",
            execute=lambda ctx: self._create_order(ctx),
            compensate=lambda ctx: self._cancel_order(ctx),
        )
        
        # Step 2: Process Payment
        saga.add_step(
            "process_payment",
            execute=lambda ctx: self._process_payment(ctx),
            compensate=lambda ctx: self._refund_payment(ctx),
        )
        
        # Step 3: Reserve Inventory
        saga.add_step(
            "reserve_inventory",
            execute=lambda ctx: self._reserve_inventory(ctx),
            compensate=lambda ctx: self._release_inventory(ctx),
        )
        
        # Step 4: Send Confirmation (no compensation needed)
        saga.add_step(
            "send_confirmation",
            execute=lambda ctx: self._send_confirmation(ctx),
            compensate=lambda ctx: SagaStepResult(success=True),  # No-op
        )
        
        return await saga.execute(order_data)
    
    async def _create_order(self, ctx: dict) -> SagaStepResult:
        order = await self.order_service.create(ctx)
        return SagaStepResult(success=True, data={"order_id": order.id})
    
    async def _cancel_order(self, ctx: dict) -> SagaStepResult:
        await self.order_service.cancel(ctx["order_id"])
        return SagaStepResult(success=True)
    
    async def _process_payment(self, ctx: dict) -> SagaStepResult:
        try:
            payment = await self.payment_service.charge(
                order_id=ctx["order_id"],
                amount=ctx["total"],
                customer_id=ctx["customer_id"],
            )
            return SagaStepResult(success=True, data={"payment_id": payment.id})
        except PaymentDeclinedError as e:
            return SagaStepResult(success=False, error=str(e))
    
    async def _refund_payment(self, ctx: dict) -> SagaStepResult:
        await self.payment_service.refund(ctx["payment_id"])
        return SagaStepResult(success=True)
    
    async def _reserve_inventory(self, ctx: dict) -> SagaStepResult:
        try:
            await self.inventory_service.reserve(
                order_id=ctx["order_id"],
                items=ctx["items"],
            )
            return SagaStepResult(success=True)
        except InsufficientStockError as e:
            return SagaStepResult(success=False, error=str(e))
    
    async def _release_inventory(self, ctx: dict) -> SagaStepResult:
        await self.inventory_service.release(ctx["order_id"])
        return SagaStepResult(success=True)
    
    async def _send_confirmation(self, ctx: dict) -> SagaStepResult:
        await self.notification_service.send_order_confirmation(ctx["order_id"])
        return SagaStepResult(success=True)

Event Sourcing

Core Concepts

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Generic, TypeVar
import json


@dataclass
class Event:
    """Base event class."""
    event_id: str
    event_type: str
    aggregate_id: str
    aggregate_type: str
    version: int
    timestamp: datetime
    data: dict
    metadata: dict = field(default_factory=dict)


T = TypeVar("T")


class Aggregate(ABC, Generic[T]):
    """Base aggregate with event sourcing."""
    
    def __init__(self, id: str):
        self.id = id
        self.version = 0
        self._changes: list[Event] = []
    
    @abstractmethod
    def apply(self, event: Event) -> None:
        """Apply event to update state."""
        pass
    
    def load_from_history(self, events: list[Event]) -> None:
        """Rebuild state from events."""
        for event in events:
            self.apply(event)
            self.version = event.version
    
    def raise_event(self, event_type: str, data: dict) -> None:
        """Raise a new event."""
        event = Event(
            event_id=str(uuid.uuid4()),
            event_type=event_type,
            aggregate_id=self.id,
            aggregate_type=self.__class__.__name__,
            version=self.version + 1,
            timestamp=datetime.utcnow(),
            data=data,
        )
        self._changes.append(event)
        self.apply(event)
        self.version = event.version
    
    def get_uncommitted_changes(self) -> list[Event]:
        return self._changes.copy()
    
    def mark_changes_committed(self) -> None:
        self._changes.clear()


# Order aggregate with event sourcing
class Order(Aggregate):
    def __init__(self, id: str):
        super().__init__(id)
        self.customer_id: str = ""
        self.items: list[dict] = []
        self.total: float = 0
        self.status: str = "draft"
    
    # Commands (business operations)
    def create(self, customer_id: str, items: list[dict]):
        if self.status != "draft":
            raise InvalidOperationError("Order already created")
        
        total = sum(item["price"] * item["quantity"] for item in items)
        
        self.raise_event("OrderCreated", {
            "customer_id": customer_id,
            "items": items,
            "total": total,
        })
    
    def confirm(self):
        if self.status != "pending":
            raise InvalidOperationError("Order cannot be confirmed")
        
        self.raise_event("OrderConfirmed", {})
    
    def cancel(self, reason: str):
        if self.status in ["shipped", "delivered"]:
            raise InvalidOperationError("Order cannot be cancelled")
        
        self.raise_event("OrderCancelled", {"reason": reason})
    
    def ship(self, tracking_number: str):
        if self.status != "confirmed":
            raise InvalidOperationError("Order must be confirmed before shipping")
        
        self.raise_event("OrderShipped", {"tracking_number": tracking_number})
    
    # Event handlers (state updates)
    def apply(self, event: Event) -> None:
        handler = getattr(self, f"_apply_{event.event_type}", None)
        if handler:
            handler(event)
    
    def _apply_OrderCreated(self, event: Event):
        self.customer_id = event.data["customer_id"]
        self.items = event.data["items"]
        self.total = event.data["total"]
        self.status = "pending"
    
    def _apply_OrderConfirmed(self, event: Event):
        self.status = "confirmed"
    
    def _apply_OrderCancelled(self, event: Event):
        self.status = "cancelled"
    
    def _apply_OrderShipped(self, event: Event):
        self.status = "shipped"

Event Store

from abc import ABC, abstractmethod


class EventStore(ABC):
    """Abstract event store."""
    
    @abstractmethod
    async def save_events(
        self, 
        aggregate_id: str,
        events: list[Event],
        expected_version: int,
    ) -> None: ...
    
    @abstractmethod
    async def get_events(
        self,
        aggregate_id: str,
        start_version: int = 0,
    ) -> list[Event]: ...


class PostgresEventStore(EventStore):
    """PostgreSQL-based event store."""
    
    def __init__(self, connection_pool):
        self.pool = connection_pool
    
    async def save_events(
        self,
        aggregate_id: str,
        events: list[Event],
        expected_version: int,
    ) -> None:
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # Optimistic concurrency check
                result = await conn.fetchval(
                    """
                    SELECT MAX(version) FROM events 
                    WHERE aggregate_id = $1
                    """,
                    aggregate_id,
                )
                current_version = result or 0
                
                if current_version != expected_version:
                    raise ConcurrencyError(
                        f"Expected version {expected_version}, "
                        f"but found {current_version}"
                    )
                
                # Insert events
                for event in events:
                    await conn.execute(
                        """
                        INSERT INTO events 
                        (event_id, aggregate_id, aggregate_type, event_type, 
                         version, timestamp, data, metadata)
                        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                        """,
                        event.event_id,
                        event.aggregate_id,
                        event.aggregate_type,
                        event.event_type,
                        event.version,
                        event.timestamp,
                        json.dumps(event.data),
                        json.dumps(event.metadata),
                    )
    
    async def get_events(
        self,
        aggregate_id: str,
        start_version: int = 0,
    ) -> list[Event]:
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT * FROM events 
                WHERE aggregate_id = $1 AND version > $2
                ORDER BY version ASC
                """,
                aggregate_id,
                start_version,
            )
            
            return [
                Event(
                    event_id=row["event_id"],
                    event_type=row["event_type"],
                    aggregate_id=row["aggregate_id"],
                    aggregate_type=row["aggregate_type"],
                    version=row["version"],
                    timestamp=row["timestamp"],
                    data=json.loads(row["data"]),
                    metadata=json.loads(row["metadata"]),
                )
                for row in rows
            ]


# Repository using event store
class EventSourcedRepository:
    def __init__(self, event_store: EventStore, aggregate_class: type):
        self.event_store = event_store
        self.aggregate_class = aggregate_class
    
    async def get(self, aggregate_id: str) -> Aggregate:
        events = await self.event_store.get_events(aggregate_id)
        
        if not events:
            raise AggregateNotFoundError(aggregate_id)
        
        aggregate = self.aggregate_class(aggregate_id)
        aggregate.load_from_history(events)
        
        return aggregate
    
    async def save(self, aggregate: Aggregate) -> None:
        changes = aggregate.get_uncommitted_changes()
        
        if not changes:
            return
        
        expected_version = aggregate.version - len(changes)
        
        await self.event_store.save_events(
            aggregate.id,
            changes,
            expected_version,
        )
        
        aggregate.mark_changes_committed()

Snapshots for Performance

class SnapshotStore:
    """Store aggregate snapshots for performance."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.snapshot_frequency = 100  # Snapshot every 100 events
    
    async def save_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str,
        version: int,
        state: dict,
    ) -> None:
        await self.redis.hset(
            f"snapshot:{aggregate_type}:{aggregate_id}",
            mapping={
                "version": str(version),
                "state": json.dumps(state),
                "timestamp": datetime.utcnow().isoformat(),
            },
        )
    
    async def get_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str,
    ) -> tuple[int, dict] | None:
        data = await self.redis.hgetall(
            f"snapshot:{aggregate_type}:{aggregate_id}"
        )
        
        if not data:
            return None
        
        return (
            int(data[b"version"]),
            json.loads(data[b"state"]),
        )


class SnapshotAwareRepository:
    """Repository that uses snapshots for faster loading."""
    
    def __init__(
        self,
        event_store: EventStore,
        snapshot_store: SnapshotStore,
        aggregate_class: type,
    ):
        self.event_store = event_store
        self.snapshot_store = snapshot_store
        self.aggregate_class = aggregate_class
    
    async def get(self, aggregate_id: str) -> Aggregate:
        # Try to load from snapshot
        snapshot = await self.snapshot_store.get_snapshot(
            aggregate_id,
            self.aggregate_class.__name__,
        )
        
        start_version = 0
        aggregate = self.aggregate_class(aggregate_id)
        
        if snapshot:
            start_version, state = snapshot
            aggregate.restore_from_snapshot(state)
            aggregate.version = start_version
        
        # Load events after snapshot
        events = await self.event_store.get_events(aggregate_id, start_version)
        
        if not events and not snapshot:
            raise AggregateNotFoundError(aggregate_id)
        
        aggregate.load_from_history(events)
        
        return aggregate
    
    async def save(self, aggregate: Aggregate) -> None:
        changes = aggregate.get_uncommitted_changes()
        
        if not changes:
            return
        
        expected_version = aggregate.version - len(changes)
        
        await self.event_store.save_events(
            aggregate.id,
            changes,
            expected_version,
        )
        
        # Create snapshot if needed
        if aggregate.version % self.snapshot_store.snapshot_frequency == 0:
            await self.snapshot_store.save_snapshot(
                aggregate.id,
                aggregate.__class__.__name__,
                aggregate.version,
                aggregate.to_snapshot(),
            )
        
        aggregate.mark_changes_committed()

CQRS (Command Query Responsibility Segregation)

from abc import ABC, abstractmethod


# Commands
@dataclass
class Command(ABC):
    pass


@dataclass
class CreateOrderCommand(Command):
    customer_id: str
    items: list[dict]


@dataclass
class ConfirmOrderCommand(Command):
    order_id: str


# Command handlers
class CommandHandler(ABC):
    @abstractmethod
    async def handle(self, command: Command) -> Any: ...


class CreateOrderHandler(CommandHandler):
    def __init__(self, repository: EventSourcedRepository):
        self.repository = repository
    
    async def handle(self, command: CreateOrderCommand) -> str:
        order = Order(str(uuid.uuid4()))
        order.create(command.customer_id, command.items)
        await self.repository.save(order)
        return order.id


class ConfirmOrderHandler(CommandHandler):
    def __init__(self, repository: EventSourcedRepository):
        self.repository = repository
    
    async def handle(self, command: ConfirmOrderCommand) -> None:
        order = await self.repository.get(command.order_id)
        order.confirm()
        await self.repository.save(order)


# Read models
class OrderReadModel:
    """Denormalized read model for orders."""
    __tablename__ = "order_views"
    
    id = Column(String, primary_key=True)
    customer_id = Column(String)
    customer_name = Column(String)  # Denormalized
    status = Column(String)
    total = Column(Float)
    item_count = Column(Integer)
    created_at = Column(DateTime)
    confirmed_at = Column(DateTime, nullable=True)
    shipped_at = Column(DateTime, nullable=True)


# Event handlers for read model
class OrderReadModelProjector:
    """Project events to read model."""
    
    def __init__(self, session, user_client):
        self.session = session
        self.user_client = user_client
    
    async def handle_event(self, event: Event):
        handler = getattr(self, f"on_{event.event_type}", None)
        if handler:
            await handler(event)
    
    async def on_OrderCreated(self, event: Event):
        # Fetch customer name for denormalization
        user = await self.user_client.get_user(event.data["customer_id"])
        
        view = OrderReadModel(
            id=event.aggregate_id,
            customer_id=event.data["customer_id"],
            customer_name=user.name,
            status="pending",
            total=event.data["total"],
            item_count=len(event.data["items"]),
            created_at=event.timestamp,
        )
        self.session.add(view)
        await self.session.commit()
    
    async def on_OrderConfirmed(self, event: Event):
        await self.session.execute(
            update(OrderReadModel)
            .where(OrderReadModel.id == event.aggregate_id)
            .values(status="confirmed", confirmed_at=event.timestamp)
        )
        await self.session.commit()
    
    async def on_OrderShipped(self, event: Event):
        await self.session.execute(
            update(OrderReadModel)
            .where(OrderReadModel.id == event.aggregate_id)
            .values(status="shipped", shipped_at=event.timestamp)
        )
        await self.session.commit()


# Queries
@dataclass
class Query(ABC):
    pass


@dataclass
class GetOrderQuery(Query):
    order_id: str


@dataclass
class ListOrdersQuery(Query):
    customer_id: str | None = None
    status: str | None = None
    limit: int = 20
    offset: int = 0


class OrderQueryHandler:
    """Handle order queries using read model."""
    
    def __init__(self, session):
        self.session = session
    
    async def get_order(self, query: GetOrderQuery) -> OrderReadModel:
        result = await self.session.execute(
            select(OrderReadModel).where(OrderReadModel.id == query.order_id)
        )
        return result.scalar_one_or_none()
    
    async def list_orders(self, query: ListOrdersQuery) -> list[OrderReadModel]:
        stmt = select(OrderReadModel)
        
        if query.customer_id:
            stmt = stmt.where(OrderReadModel.customer_id == query.customer_id)
        if query.status:
            stmt = stmt.where(OrderReadModel.status == query.status)
        
        stmt = stmt.limit(query.limit).offset(query.offset)
        
        result = await self.session.execute(stmt)
        return result.scalars().all()

Outbox Pattern

# Ensure events are published reliably
class OutboxEntry:
    __tablename__ = "outbox"
    
    id = Column(String, primary_key=True)
    aggregate_type = Column(String, nullable=False)
    aggregate_id = Column(String, nullable=False)
    event_type = Column(String, nullable=False)
    payload = Column(JSON, nullable=False)
    created_at = Column(DateTime, nullable=False)
    published_at = Column(DateTime, nullable=True)


class OutboxRepository:
    """Save events to outbox table."""
    
    async def save_with_outbox(
        self,
        session,
        aggregate: Aggregate,
    ) -> None:
        """Save aggregate and outbox entries in same transaction."""
        changes = aggregate.get_uncommitted_changes()
        
        for event in changes:
            outbox_entry = OutboxEntry(
                id=event.event_id,
                aggregate_type=event.aggregate_type,
                aggregate_id=event.aggregate_id,
                event_type=event.event_type,
                payload={
                    "event_id": event.event_id,
                    "data": event.data,
                    "metadata": event.metadata,
                    "timestamp": event.timestamp.isoformat(),
                },
                created_at=datetime.utcnow(),
            )
            session.add(outbox_entry)
        
        await session.commit()
        aggregate.mark_changes_committed()


class OutboxProcessor:
    """Background processor to publish outbox events."""
    
    def __init__(self, session_factory, event_publisher, batch_size: int = 100):
        self.session_factory = session_factory
        self.event_publisher = event_publisher
        self.batch_size = batch_size
    
    async def process(self):
        """Process pending outbox entries."""
        async with self.session_factory() as session:
            # Get unpublished entries
            entries = await session.execute(
                select(OutboxEntry)
                .where(OutboxEntry.published_at.is_(None))
                .limit(self.batch_size)
                .with_for_update(skip_locked=True)
            )
            entries = entries.scalars().all()
            
            for entry in entries:
                try:
                    await self.event_publisher.publish(
                        entry.event_type,
                        entry.payload,
                    )
                    entry.published_at = datetime.utcnow()
                except Exception as e:
                    logger.error(f"Failed to publish {entry.id}: {e}")
            
            await session.commit()
    
    async def run_forever(self, interval: float = 1.0):
        """Continuously process outbox."""
        while True:
            await self.process()
            await asyncio.sleep(interval)

Key Takeaways

1. Sagas for distributed transactions - Choreography for simple flows, orchestration for complex ones

2. Event sourcing for audit - Store events, derive state

3. CQRS for optimization - Separate read and write models

4. Eventual consistency is normal - Design for it from the start

5. Outbox for reliability - Ensure events are published

What's Next?

Distributed systems fail. In Article 9: Resilience Patterns, we'll cover circuit breakers, retries, bulkheads, and fallback strategies to build fault-tolerant services.

---

This article is part of the Microservice Architecture 101 series.