What Is the Unit of Work Pattern? Transactions and Consistency Explained

The Unit of Work Pattern coordinates changes to multiple objects and repositories within a single transactional boundary. Instead of allowing every repository to commit independently, a Unit of Work groups related operations and determines when their changes should be persisted together. This helps prevent partially completed business operations, simplifies transaction ownership, and makes application behavior easier to test.

Unit of Work is closely related to the Repository Pattern and is described by Martin Fowler in Patterns of Enterprise Application Architecture. Many modern ORMs, including SQLAlchemy, already implement change tracking and transaction coordination internally. An application-level Unit of Work adds a useful architectural boundary around those capabilities. This guide explains the pattern, its relationship with transactions and repositories, common concurrency problems, and a practical Python/FastAPI/PostgreSQL implementation that creates an order, reserves inventory, and writes a transactional outbox event atomically.

Table of Contents

• What Is the Unit of Work Pattern?
  • The Problem Unit of Work Solves
  • Core Responsibilities and Transaction Lifecycle
• How the Unit of Work Pattern Works
  • Change Tracking, Flush, and Commit
  • Transaction Boundaries and Rollback
  • Repository vs. Unit of Work vs. ORM Session
• Transactions, Isolation, and Concurrency
  • ACID Properties and Database Guarantees
  • Optimistic Locking and Lost Updates
  • Why One Unit of Work Cannot Guarantee Distributed Atomicity
• Production Design Example: Order Placement
  • Business Requirements and Repository Responsibilities
  • Atomic Order, Inventory, and Outbox Updates
  • Failure Scenarios and Expected Outcomes
• Ready-to-Use Example with FastAPI and PostgreSQL
  • Project Structure and Dependencies
  • Domain Models and Business Invariants
  • Application Use Cases and Unit of Work Contract
  • PostgreSQL Repositories and Transaction Implementation
  • FastAPI Adapter and Dependency Injection
  • Unit Tests for Commit and Rollback
  • Run the Application and Verify Database Consistency
• When to Use the Unit of Work Pattern
  • Where Unit of Work Provides Value
  • When an Explicit Unit of Work Is Unnecessary
• Common Mistakes and Production Practices
  • Common Implementation Mistakes
  • Production Checklist
• Frequently Asked Questions
  • Does SQLAlchemy Already Implement Unit of Work?
  • What Is the Difference Between Repository and Unit of Work?
  • Should Every HTTP Request Have a Unit of Work?
  • Can a Unit of Work Contain Nested Transactions?
  • Does Unit of Work Work Across Microservices?
  • Should Repositories Call Commit?
• Conclusion
  • More Articles to Read

What Is the Unit of Work Pattern?

A Unit of Work represents a set of related changes that should be persisted as one operation. It coordinates the persistence of new, modified, and deleted objects and handles the transaction boundary around those changes.

In a business application, one user action may require updates to several database tables or aggregates. The Unit of Work ensures those changes can be coordinated within a single database transaction rather than being committed independently by each repository.

The Problem Unit of Work Solves

Consider an order placement operation. The application creates an order, decreases available inventory, and records an event for downstream processing.

A naive implementation might look like this:

def place_order(order, inventory):
    # Each repository commits independently.
    order_repository.add_and_commit(order)

    inventory_repository.reserve_and_commit(
        inventory.sku,
        order.quantity,
    )

    outbox_repository.add_and_commit(
        "OrderPlaced",
        {"order_id": order.order_id},
    )

If inventory reservation fails after the order is committed, the database contains an order without the required stock reservation.

If event recording fails after the first two commits, the business state changes but no durable event is available for downstream consumers.

These partial failures are difficult to repair automatically because the individual commits have already completed.

Unit of Work changes transaction ownership. Repositories perform their operations within a shared transaction, and the application commits only after all required operations succeed.

Core Responsibilities and Transaction Lifecycle

An application-level Unit of Work commonly provides:

• Transaction scope: Defines where one logical database transaction begins and ends.
• Shared persistence context: Allows multiple repositories to use the same database session.
• Commit coordination: Persists all required changes together.
• Rollback: Discards uncommitted changes when the operation fails.
• Resource cleanup: Releases sessions and database connections.
• Application independence: Prevents business use cases from depending directly on SQLAlchemy transaction APIs.

Martin Fowler's original description also emphasizes tracking changed objects and coordinating how those changes are written. In a modern ORM application, the ORM's change tracker often performs that lower-level responsibility.

The application-level Unit of Work therefore does not need to reimplement ORM change tracking. It can delegate persistence mechanics to SQLAlchemy while making transaction scope explicit for the application.

How the Unit of Work Pattern Works

A practical Unit of Work creates or owns a persistence context, exposes the repositories participating in the transaction, and provides an explicit commit operation.

Its lifecycle should be predictable. Starting a new business operation creates a new transactional scope, while completion either commits or rolls back the changes.

Change Tracking, Flush, and Commit

ORMs such as SQLAlchemy track objects associated with a Session. New objects can be scheduled for insertion, while modifications to mapped objects can be detected and synchronized with the database.

Three concepts should be distinguished:

• Pending changes: Modifications recorded in application memory or the persistence session.
• Flush: Sends pending SQL statements to the database within the active transaction.
• Commit: Requests that the database make the transaction durable and visible according to its isolation rules.

A flush is not equivalent to a commit. The database may have executed an INSERT or UPDATE, but the transaction can still roll back.

For example, SQLAlchemy may flush an order insertion before running a dependent query. If an exception occurs before commit, the entire transaction can still be rolled back.

This distinction is important when debugging applications that observe SQL statements executing but do not see the expected records after an error.

Transaction Boundaries and Rollback

Transaction scope should follow the business operation rather than the implementation details of individual repositories.

A typical application-level interface looks like this:

# Conceptual usage:
#
# UnitOfWork, CreateOrder, and InventoryReservation
# are application-defined components.

with unit_of_work_factory() as uow:
    uow.orders.add(order)
    uow.inventory.reserve(sku, quantity)
    uow.outbox.add(event)

    # Commit the three changes together.
    uow.commit()

# Without commit, the context rolls back.

If the inventory repository raises an exception, the application leaves the transaction without committing it. The Unit of Work rolls back any earlier uncommitted changes.

The exact implementation matters. A Unit of Work should not automatically commit every time its context manager exits successfully unless that behavior is an explicit part of its contract.

An explicit-commit design makes it clear which application operation authorizes durable changes.

Repository vs. Unit of Work vs. ORM Session

These concepts solve overlapping but different problems.

A Repository hides persistence details behind domain-oriented operations. A Unit of Work coordinates several of those operations.

SQLAlchemy Session already implements many Unit of Work concepts internally. A custom wrapper is valuable when it protects an application boundary and prevents use cases from depending on ORM-specific transaction APIs.

Transactions, Isolation, and Concurrency

Unit of Work organizes transaction coordination, but the database is responsible for enforcing actual transaction guarantees. Wrapping several repository calls in a Python class does not by itself make them atomic.

Correctness also depends on isolation levels, constraints, conditional updates, and how concurrent operations interact.

ACID Properties and Database Guarantees

Relational transactions are commonly described using ACID properties:

• Atomicity: All operations in the transaction commit or none of them do.
• Consistency: Database constraints and application rules maintain valid state.
• Isolation: Concurrent transactions interact according to the configured isolation level.
• Durability: Successfully committed changes survive failures according to the database's durability guarantees.

Unit of Work helps the application select which operations should share one transaction. It does not replace primary-key constraints, check constraints, or safe concurrent update logic.

For example, two concurrent orders might both try to reserve the final available item. A transaction alone does not make a separate read-then-write sequence safe under every isolation level.

The inventory update must enforce that the available quantity remains sufficient, typically through locking or an atomic conditional update.

Optimistic Locking and Lost Updates

Optimistic concurrency control is commonly used when competing updates are possible but collisions are not expected to be constant.

An aggregate includes a version field, and updates are conditioned on the version previously observed by the application.

For example:

-- Update only if the caller read version 4.
UPDATE orders
SET
    status = 'confirmed',
    version = version + 1
WHERE
    order_id = 'order-1001'
    AND version = 4
RETURNING version;

If the UPDATE returns no rows, the application knows that the expected state was not present. It must distinguish a stale version from a missing or otherwise ineligible record if its API requires that distinction.

For inventory reservations, a conditional decrement can enforce a different invariant:

-- Prevent reservations that exceed stock.
UPDATE inventory
SET available = available - 2
WHERE
    sku = 'keyboard'
    AND available >= 2
RETURNING available;

The database evaluates the condition as part of the update. This is safer than separately reading available inventory and later issuing an unconditional decrement.

Unit of Work can coordinate the resulting inventory change with order creation and event recording within the same transaction.

Why One Unit of Work Cannot Guarantee Distributed Atomicity

One Unit of Work over one PostgreSQL transaction can atomically update several tables in that database.

It cannot automatically commit PostgreSQL, DynamoDB, and an external payment provider as one atomic operation.

If application code sends a message to SQS inside a PostgreSQL transaction, the message may be delivered even when the database transaction later rolls back.

If the application commits PostgreSQL and then publishes the message, the process may crash between those two operations.

The Transactional Outbox Pattern addresses this database-and-message gap by recording outgoing events in the same local transaction as the business change.

A separate publisher then sends the events to the broker. This provides durable handoff but still requires duplicate-safe delivery and consumption.

For business workflows spanning independently owned services, a Saga or another distributed coordination approach may be appropriate. Unit of Work remains useful inside each local transaction.

Production Design Example: Order Placement

Consider an order API serving an e-commerce application. A customer submits an order for two units of a product.

The application must create the order, reserve the units from available inventory, and record an OrderPlaced event for downstream processing.

Business Requirements and Repository Responsibilities

The system must satisfy the following requirements:

• Order IDs are unique.
• Requested quantities must be positive.
• Inventory must not become negative.
• No order should be committed without its inventory reservation.
• No committed order should lack its corresponding durable outbox event.
• A failed reservation must leave all related tables unchanged.
• Application tests must be able to simulate commit and rollback without PostgreSQL.

Three repositories participate in the operation:

• OrderRepository: Creates the Order record.
• InventoryRepository: Atomically decreases available inventory.
• OutboxRepository: Records an event for later publication.

The Unit of Work supplies one shared PostgreSQL Session to all three repositories and owns the transaction boundary.

Atomic Order, Inventory, and Outbox Updates

The intended workflow is:

1. Start a Unit of Work.
2. Insert the pending order.
3. Reserve the requested inventory with an atomic conditional update.
4. Insert an OrderPlaced event into the outbox table.
5. Commit the transaction once all operations succeed.

The order and outbox cannot be partially committed if they use the same PostgreSQL transaction.

Importantly, the outbox record does not mean the event has already reached a message broker. It means the business transaction durably recorded the event for a separate publisher.

Failure Scenarios and Expected Outcomes

Consider these outcomes:

The last case illustrates the difference between a successful local transaction and successful distributed event delivery.

That distinction should be reflected in code, monitoring, and operational recovery procedures.

Ready-to-Use Example with FastAPI and PostgreSQL

This implementation uses Python 3.12 or newer, FastAPI, SQLAlchemy 2, PostgreSQL, and pytest. Its main purpose is to show how multiple repositories share one database transaction through a Unit of Work.

The same application use case can run against an in-memory test Unit of Work, allowing rollback scenarios to be tested without starting a database.

Project Structure and Dependencies

Create the following structure, including empty __init__.py files in the Python package directories:

unit-of-work-orders/
  app/
    __init__.py
    domain.py
    use_cases.py
    persistence.py
    api.py
  tests/
    test_orders.py
  requirements.txt
  docker-compose.yml

Create requirements.txt:

fastapi>=0.115,<1
uvicorn>=0.30,<1
sqlalchemy>=2.0,<3
psycopg[binary]>=3.2,<4
pytest>=8,<10
httpx>=0.27,<1

For production, lock tested dependency versions and use versioned database migrations. The application below uses startup-time table creation only to simplify local execution.

Domain Models and Business Invariants

First, define the Order entity and the OrderPlaced domain event. These domain objects contain no SQLAlchemy or FastAPI imports.

Create app/domain.py:

from dataclasses import dataclass
from datetime import datetime, timezone
from uuid import uuid4


@dataclass(frozen=True)
class Order:
    order_id: str
    customer_id: str
    sku: str
    quantity: int
    amount_cents: int
    status: str = "pending"

    def __post_init__(self) -> None:
        if not self.order_id.strip():
            raise ValueError("Order ID is required")

        if not self.customer_id.strip():
            raise ValueError("Customer ID is required")

        if not self.sku.strip():
            raise ValueError("SKU is required")

        if type(self.quantity) is not int or self.quantity <= 0:
            raise ValueError("Quantity must be positive")

        if (
            type(self.amount_cents) is not int
            or self.amount_cents <= 0
        ):
            raise ValueError("Amount must be positive")


@dataclass(frozen=True)
class OrderPlaced:
    event_id: str
    order_id: str
    occurred_at: datetime

    @classmethod
    def create(cls, order_id: str) -> "OrderPlaced":
        return cls(
            event_id=str(uuid4()),
            order_id=order_id,
            occurred_at=datetime.now(timezone.utc),
        )

The domain validates fundamental business state, while the application use case will coordinate persistence and inventory reservation.

The amount is expressed in integer cents. The example assumes a single currency and intentionally excludes pricing, taxes, and payment authorization.

Application Use Cases and Unit of Work Contract

The application layer defines repository capabilities and the Unit of Work interface through Python Protocols.

Each repository handles its own domain-oriented persistence operation, but none of them exposes a commit method.

Create app/use_cases.py:

from dataclasses import dataclass
from typing import Callable, Protocol, Self

from app.domain import Order, OrderPlaced


class OrderAlreadyExists(Exception):
    pass


class InsufficientInventory(Exception):
    pass


class OrderRepository(Protocol):
    def add(self, order: Order) -> bool:
        """Return False when order ID already exists."""

    def get(self, order_id: str) -> Order | None:
        """Retrieve one order."""


class InventoryRepository(Protocol):
    def reserve(
        self,
        sku: str,
        quantity: int,
    ) -> bool:
        """Atomically reserve stock if available."""


class OutboxRepository(Protocol):
    def add(self, event: OrderPlaced) -> None:
        """Record a pending event in this transaction."""


class UnitOfWork(Protocol):
    orders: OrderRepository
    inventory: InventoryRepository
    outbox: OutboxRepository

    def __enter__(self) -> Self:
        ...

    def __exit__(self, exc_type, exc, tb) -> None:
        ...

    def commit(self) -> None:
        ...


@dataclass(frozen=True)
class PlaceOrderCommand:
    order_id: str
    customer_id: str
    sku: str
    quantity: int
    amount_cents: int


@dataclass(frozen=True)
class OrderResult:
    order_id: str
    status: str


class PlaceOrder:
    def __init__(
        self,
        uow_factory: Callable[[], UnitOfWork],
    ) -> None:
        self.uow_factory = uow_factory

    def execute(
        self,
        command: PlaceOrderCommand,
    ) -> OrderResult:
        order = Order(
            order_id=command.order_id,
            customer_id=command.customer_id,
            sku=command.sku,
            quantity=command.quantity,
            amount_cents=command.amount_cents,
        )

        with self.uow_factory() as uow:
            # No individual repository commits.
            if not uow.orders.add(order):
                raise OrderAlreadyExists(order.order_id)

            if not uow.inventory.reserve(
                order.sku,
                order.quantity,
            ):
                raise InsufficientInventory(order.sku)

            # Persist event together with business state.
            uow.outbox.add(
                OrderPlaced.create(order.order_id)
            )

            # One commit for the complete operation.
            uow.commit()

        return OrderResult(
            order_id=order.order_id,
            status=order.status,
        )

The use case defines exactly which changes must succeed together. If order creation, inventory reservation, or event recording fails, the context exits without a successful commit.

It also keeps transaction management independent of SQLAlchemy. The same use case can receive a PostgreSQL Unit of Work or a test implementation.

The explicit commit happens only after all three repository operations have completed.

PostgreSQL Repositories and Transaction Implementation

The infrastructure layer implements three repositories backed by one SQLAlchemy Session. The Unit of Work owns the Session and the database transaction.

Create app/persistence.py:

import os
from datetime import datetime

from sqlalchemy import (
    DateTime,
    Integer,
    JSON,
    String,
    create_engine,
    update,
)
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    sessionmaker,
)

from app.domain import Order, OrderPlaced


DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg://app:app@localhost:5432/orders",
)

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)

SessionFactory = sessionmaker(
    bind=engine,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


class OrderRow(Base):
    __tablename__ = "orders"

    order_id: Mapped[str] = mapped_column(
        String(100),
        primary_key=True,
    )
    customer_id: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )
    sku: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )
    quantity: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )
    amount_cents: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(30),
        nullable=False,
    )


class InventoryRow(Base):
    __tablename__ = "inventory"

    sku: Mapped[str] = mapped_column(
        String(100),
        primary_key=True,
    )
    available: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )


class OutboxRow(Base):
    __tablename__ = "outbox"

    event_id: Mapped[str] = mapped_column(
        String(100),
        primary_key=True,
    )
    aggregate_id: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )
    event_type: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )
    payload: Mapped[dict] = mapped_column(
        JSON,
        nullable=False,
    )
    occurred_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
    )


class SqlOrderRepository:
    def __init__(self, session: Session):
        self.session = session

    def add(self, order: Order) -> bool:
        statement = (
            insert(OrderRow)
            .values(
                order_id=order.order_id,
                customer_id=order.customer_id,
                sku=order.sku,
                quantity=order.quantity,
                amount_cents=order.amount_cents,
                status=order.status,
            )
            .on_conflict_do_nothing(
                index_elements=["order_id"]
            )
            .returning(OrderRow.order_id)
        )

        created_id = self.session.execute(
            statement
        ).scalar_one_or_none()

        return created_id is not None

    def get(self, order_id: str) -> Order | None:
        row = self.session.get(OrderRow, order_id)

        if row is None:
            return None

        return Order(
            order_id=row.order_id,
            customer_id=row.customer_id,
            sku=row.sku,
            quantity=row.quantity,
            amount_cents=row.amount_cents,
            status=row.status,
        )


class SqlInventoryRepository:
    def __init__(self, session: Session):
        self.session = session

    def reserve(
        self,
        sku: str,
        quantity: int,
    ) -> bool:
        # Conditional update prevents negative stock
        # even with concurrent order requests.
        statement = (
            update(InventoryRow)
            .where(
                InventoryRow.sku == sku,
                InventoryRow.available >= quantity,
            )
            .values(
                available=(
                    InventoryRow.available - quantity
                )
            )
            .returning(InventoryRow.sku)
        )

        reserved = self.session.execute(
            statement
        ).scalar_one_or_none()

        return reserved is not None


class SqlOutboxRepository:
    def __init__(self, session: Session):
        self.session = session

    def add(self, event: OrderPlaced) -> None:
        self.session.add(
            OutboxRow(
                event_id=event.event_id,
                aggregate_id=event.order_id,
                event_type="OrderPlaced",
                payload={
                    "event_id": event.event_id,
                    "order_id": event.order_id,
                },
                occurred_at=event.occurred_at,
            )
        )


class SqlAlchemyUnitOfWork:
    def __init__(self, session_factory=SessionFactory):
        self.session_factory = session_factory

    def __enter__(self):
        self.session = self.session_factory()
        self.transaction = self.session.begin()
        self.committed = False

        # All repositories share one Session.
        self.orders = SqlOrderRepository(self.session)
        self.inventory = SqlInventoryRepository(
            self.session
        )
        self.outbox = SqlOutboxRepository(self.session)

        return self

    def commit(self) -> None:
        # Flush pending outbox rows and commit.
        self.transaction.commit()
        self.committed = True

    def __exit__(self, exc_type, exc, tb) -> None:
        try:
            if not self.committed:
                self.transaction.rollback()
        finally:
            self.session.close()

The key design decision is that all three repositories share the same SQLAlchemy Session and transaction.

The order repository uses an atomic insert with a primary-key conflict check. The inventory repository uses an atomic conditional decrement, and the outbox repository stages a new event row.

No repository calls commit(). Only the Unit of Work commits the transaction after every required operation has succeeded.

When transaction.commit() runs, SQLAlchemy flushes pending changes, including the outbox row, before issuing the database commit.

If that flush or commit fails, SQLAlchemy raises an exception. The context manager closes the Session, and the database transaction does not successfully commit. A production service should classify retryable errors and investigate ambiguous commit outcomes rather than assuming every failure is safe to retry blindly.

This implementation uses a single-use Unit of Work with one explicit commit. Reusing the same instance after commit is intentionally outside its contract.

FastAPI Adapter and Dependency Injection

The HTTP adapter converts requests into application commands and supplies a Unit of Work factory.

Create app/api.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from app.persistence import (
    Base,
    InventoryRow,
    SessionFactory,
    SqlAlchemyUnitOfWork,
    engine,
)
from app.use_cases import (
    InsufficientInventory,
    OrderAlreadyExists,
    PlaceOrder,
    PlaceOrderCommand,
)


class PlaceOrderRequest(BaseModel):
    order_id: str = Field(min_length=1)
    customer_id: str = Field(min_length=1)
    sku: str = Field(min_length=1)
    quantity: int = Field(gt=0)
    amount_cents: int = Field(gt=0)


class OrderResponse(BaseModel):
    order_id: str
    status: str


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Local demo only; production needs migrations.
    Base.metadata.create_all(bind=engine)

    # Seed inventory without resetting existing stock.
    with SessionFactory.begin() as session:
        if session.get(InventoryRow, "keyboard") is None:
            session.add(
                InventoryRow(
                    sku="keyboard",
                    available=10,
                )
            )

    yield
    engine.dispose()


app = FastAPI(
    title="Unit of Work Orders API",
    lifespan=lifespan,
)


def new_unit_of_work() -> SqlAlchemyUnitOfWork:
    return SqlAlchemyUnitOfWork(SessionFactory)


@app.post(
    "/orders",
    response_model=OrderResponse,
    status_code=201,
)
def place_order(request: PlaceOrderRequest):
    command = PlaceOrderCommand(
        order_id=request.order_id,
        customer_id=request.customer_id,
        sku=request.sku,
        quantity=request.quantity,
        amount_cents=request.amount_cents,
    )

    try:
        return PlaceOrder(
            new_unit_of_work
        ).execute(command)

    except OrderAlreadyExists as exc:
        raise HTTPException(
            status_code=409,
            detail="Order already exists",
        ) from exc

    except InsufficientInventory as exc:
        raise HTTPException(
            status_code=409,
            detail="Insufficient inventory",
        ) from exc


@app.get("/health")
def health():
    # This only checks the application process.
    # A database readiness check is separate.
    return {"status": "ok"}

The application service does not import SQLAlchemy, while the FastAPI adapter is responsible for wiring the concrete infrastructure and mapping application errors to HTTP responses.

The example seeds ten units of keyboard inventory for local testing. It does not reset stock on each application startup.

For production deployments, use versioned migrations, a separate controlled inventory-initialization process, authentication, and appropriate order ownership checks.

Unit Tests for Commit and Rollback

The application should be tested without requiring PostgreSQL for every business-rule test.

An in-memory Unit of Work can maintain a temporary copy of the persisted state. A successful commit publishes those staged changes into the fake database; leaving the context without commit discards them.

Create tests/test_orders.py:

from copy import deepcopy

import pytest

from app.domain import Order
from app.use_cases import (
    InsufficientInventory,
    OrderAlreadyExists,
    PlaceOrder,
    PlaceOrderCommand,
)


class FakeDatabase:
    def __init__(self):
        self.orders: dict[str, Order] = {}
        self.inventory = {"keyboard": 10}
        self.outbox = []


class FakeOrderRepository:
    def __init__(self, state):
        self.state = state

    def add(self, order: Order) -> bool:
        if order.order_id in self.state.orders:
            return False

        self.state.orders[order.order_id] = order
        return True

    def get(self, order_id: str):
        return self.state.orders.get(order_id)


class FakeInventoryRepository:
    def __init__(self, state):
        self.state = state

    def reserve(self, sku: str, quantity: int):
        available = self.state.inventory.get(
            sku,
            0,
        )

        if available < quantity:
            return False

        self.state.inventory[sku] = (
            available - quantity
        )
        return True


class FakeOutboxRepository:
    def __init__(self, state):
        self.state = state

    def add(self, event):
        self.state.outbox.append(event)


class FakeUnitOfWork:
    def __init__(self, database: FakeDatabase):
        self.database = database

    def __enter__(self):
        # Stage all changes in an isolated copy.
        self.staged = deepcopy(self.database)
        self.committed = False

        self.orders = FakeOrderRepository(
            self.staged
        )
        self.inventory = FakeInventoryRepository(
            self.staged
        )
        self.outbox = FakeOutboxRepository(
            self.staged
        )
        return self

    def commit(self):
        # This fake assumes sequential execution.
        self.database.orders = self.staged.orders
        self.database.inventory = (
            self.staged.inventory
        )
        self.database.outbox = self.staged.outbox
        self.committed = True

    def __exit__(self, exc_type, exc, tb):
        # No commit means the staged copy is discarded.
        pass


def command(
    order_id="order-1001",
    quantity=2,
):
    return PlaceOrderCommand(
        order_id=order_id,
        customer_id="customer-42",
        sku="keyboard",
        quantity=quantity,
        amount_cents=4999,
    )


def test_successful_transaction():
    database = FakeDatabase()
    service = PlaceOrder(
        lambda: FakeUnitOfWork(database)
    )

    result = service.execute(command())

    assert result.status == "pending"
    assert len(database.orders) == 1
    assert database.inventory["keyboard"] == 8
    assert len(database.outbox) == 1


def test_insufficient_inventory_rolls_back():
    database = FakeDatabase()
    service = PlaceOrder(
        lambda: FakeUnitOfWork(database)
    )

    with pytest.raises(InsufficientInventory):
        service.execute(
            command(quantity=11)
        )

    # No partial order or event is persisted.
    assert database.orders == {}
    assert database.inventory["keyboard"] == 10
    assert database.outbox == []


def test_duplicate_order_does_not_reserve_again():
    database = FakeDatabase()
    service = PlaceOrder(
        lambda: FakeUnitOfWork(database)
    )

    service.execute(command())

    with pytest.raises(OrderAlreadyExists):
        service.execute(command())

    assert len(database.orders) == 1
    assert database.inventory["keyboard"] == 8
    assert len(database.outbox) == 1


def test_uncommitted_changes_are_discarded():
    database = FakeDatabase()

    with FakeUnitOfWork(database) as uow:
        order = Order(
            order_id="order-2001",
            customer_id="customer-42",
            sku="keyboard",
            quantity=2,
            amount_cents=4999,
        )

        uow.orders.add(order)
        uow.inventory.reserve("keyboard", 2)

        # Deliberately omit commit().

    assert database.orders == {}
    assert database.inventory["keyboard"] == 10


def test_invalid_quantity_changes_nothing():
    database = FakeDatabase()
    service = PlaceOrder(
        lambda: FakeUnitOfWork(database)
    )

    with pytest.raises(ValueError):
        service.execute(
            command(quantity=0)
        )

    assert database.orders == {}
    assert database.inventory["keyboard"] == 10
    assert database.outbox == []

The tests verify that application behavior treats order creation, stock reservation, and event recording as one logical operation.

The fake does not reproduce PostgreSQL concurrency, locking, or transaction isolation. It intentionally models sequential state changes for isolated unit tests.

Integration tests should exercise the real SQLAlchemy Unit of Work against PostgreSQL, including insufficient-stock failures, duplicate IDs, outbox insertion failures, and simultaneous reservations for the final available inventory.

Run the Application and Verify Database Consistency

For local development, start PostgreSQL through Docker Compose.

Create docker-compose.yml:

services:
  postgres:
    image: postgres:16-alpine

    environment:
      POSTGRES_DB: orders
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app

    ports:
      - "127.0.0.1:5432:5432"

    healthcheck:
      test:
        - CMD-SHELL
        - pg_isready -U app -d orders
      interval: 5s
      timeout: 3s
      retries: 10

Run the application and inspect the database after both successful and failed operations:

# Start PostgreSQL.
docker compose up -d --wait

# Prepare Python.
python3 -m venv .venv
source .venv/bin/activate

pip install -r requirements.txt

# Run unit tests.
pytest -q

# Configure the database.
export DATABASE_URL="postgresql+psycopg://app:app@localhost:5432/orders"

# Start the API in a terminal.
uvicorn app.api:app --reload

# In another terminal, create a valid order.
curl -i -X POST \
  http://127.0.0.1:8000/orders \
  -H "Content-Type: application/json" \
  -d '{
    "order_id": "order-1001",
    "customer_id": "customer-42",
    "sku": "keyboard",
    "quantity": 2,
    "amount_cents": 4999
  }'

# Try reserving more than the remaining stock.
curl -i -X POST \
  http://127.0.0.1:8000/orders \
  -H "Content-Type: application/json" \
  -d '{
    "order_id": "order-1002",
    "customer_id": "customer-42",
    "sku": "keyboard",
    "quantity": 20,
    "amount_cents": 4999
  }'

# Inspect orders.
docker compose exec postgres \
  psql -U app -d orders \
  -c 'SELECT order_id, sku, quantity FROM orders ORDER BY order_id;'

# Inspect inventory.
docker compose exec postgres \
  psql -U app -d orders \
  -c 'SELECT sku, available FROM inventory;'

# Inspect recorded events.
docker compose exec postgres \
  psql -U app -d orders \
  -c 'SELECT event_type, aggregate_id FROM outbox;'

# Clean up when finished.
docker compose down

For a fresh database, the successful request should create order-1001, reduce keyboard inventory from 10 to 8, and record one OrderPlaced event.

The second request should return 409 Conflict because it requests more stock than is available. It should not persist order-1002 or record a corresponding outbox event.

The database queries allow you to verify that the failed operation left no partially committed state.

For a stronger test, deliberately cause an outbox insertion failure and verify that both the order insertion and inventory decrement roll back. Also test concurrent requests against a real PostgreSQL instance rather than assuming the in-memory fake provides database-equivalent isolation.

When to Use the Unit of Work Pattern

Unit of Work is most valuable when a business operation touches multiple persistent objects or repositories and must control when the changes become durable.

It is less useful when it only adds a wrapper around an existing ORM transaction without improving clarity, testing, or dependency boundaries.

Where Unit of Work Provides Value

Good use cases include:

• Order creation and inventory reservation.
• Account balance transfers within one database.
• Updating aggregates and inserting transactional outbox events.
• Inventory adjustments across related records.
• Financial operations requiring explicit commit boundaries.
• Application services coordinating several repositories.
• Systems using Hexagonal, Onion, or Clean Architecture that isolate persistence APIs.

Unit of Work is especially useful when repositories are intentionally prohibited from committing independently.

It provides a clear architectural location for transaction policy while allowing repositories to focus on retrieving and modifying domain objects.

When an Explicit Unit of Work Is Unnecessary

SQLAlchemy already tracks changes and supports transactional context managers. A simple application can use those capabilities directly without creating an additional Unit of Work class.

For example, a small administrative endpoint that updates one row may be adequately served by:

# SQLAlchemy transaction management directly.
#
# SessionFactory is a configured SQLAlchemy
# sessionmaker, and AuditRow is an ORM model.

with SessionFactory.begin() as session:
    session.add(
        AuditRow(
            message="Configuration updated"
        )
    )

This is valid when the application does not need an additional domain-oriented transaction boundary.

A custom Unit of Work becomes worthwhile when it makes application behavior more explicit, enables isolated testing, or coordinates several repositories through one consistent contract.

Do not introduce a complex framework merely to rename session.commit() as uow.commit().

Common Mistakes and Production Practices

Unit of Work simplifies local transaction coordination but does not eliminate database contention, distributed failures, or the need for careful retry design.

Its production quality depends on how accurately the implementation preserves the transaction contract promised to application code.

Common Implementation Mistakes

Common mistakes include:

• Independent repository commits: Related updates become partially durable.
• Different database sessions per repository: The operation no longer uses one transaction.
• Committing in a finally block: Failed business operations may be committed accidentally.
• Confusing flush with commit: Application code assumes changes are already durable.
• Keeping transactions open during network calls: Database locks and connection usage increase unnecessarily.
• Assuming transactions prevent every race: Database isolation and conditional updates still matter.
• Retrying ambiguous commits blindly: A repeated operation may create duplicate business effects.
• Publishing messages before commit: External consumers may observe events for rolled-back state.
• Publishing messages immediately after commit without durable recording: Process crashes can lose events.
• Sharing SQLAlchemy Sessions across concurrent requests: Session state is not designed for unrestricted concurrent use.
• Using nested transactions without understanding savepoints: Partial rollbacks may not produce the expected application semantics.
• Assuming one Unit of Work spans independent microservices: Local database transactions cannot guarantee distributed atomicity.
• Testing only with fakes: Real transaction and concurrency behavior remains unverified.

The most important distinction is between coordinating a local database transaction and coordinating a complete distributed business workflow.

Unit of Work solves the former. Distributed consistency generally requires additional patterns and explicit recovery policies.

Production Checklist

Before using Unit of Work in production, verify that:

• Every business operation has a clearly defined transaction boundary.
• Participating repositories share the same database session and transaction.
• Repositories do not commit independently when the application owns the transaction.
• Commit and rollback behavior are explicit and predictable.
• Sessions and connections are closed even when exceptions occur.
• Database constraints enforce critical uniqueness and validity rules.
• Concurrent modifications use suitable locking or conditional updates.
• Isolation level matches the required consistency guarantees.
• Transactions do not remain open across slow external API calls unnecessarily.
• Deadlocks and serialization failures have appropriate retry handling.
• Operation IDs or idempotency keys protect important retries.
• Business changes and outbox records commit atomically when required.
• Outbox publication has monitoring, retries, and duplicate-safe consumption.
• Database migrations are version controlled.
• Unit tests cover successful commit and rollback paths.
• Integration tests verify actual database rollback and concurrency behavior.
• Metrics expose transaction failures, lock contention, and slow commits.

Frequently Asked Questions

Unit of Work is frequently discussed alongside the Repository Pattern, ORM Sessions, database transactions, and distributed consistency patterns. The distinction between these concepts determines whether an application needs an explicit Unit of Work implementation.

Does SQLAlchemy Already Implement Unit of Work?

Yes, SQLAlchemy's Session provides change tracking and transaction-management capabilities associated with the Unit of Work pattern.

An additional application-level Unit of Work can encapsulate that Session and expose a domain-oriented transaction contract to application use cases.

It is not necessary for every SQLAlchemy application.

What Is the Difference Between Repository and Unit of Work?

A Repository provides operations for retrieving and persisting domain objects. A Unit of Work coordinates the transaction containing one or more repository operations.

Repositories answer how the application accesses domain state. Unit of Work answers when a group of changes is committed or rolled back.

Should Every HTTP Request Have a Unit of Work?

Not necessarily. Transaction boundaries should reflect the application operation.

A read-only request may not need an explicit commit. A request containing several related writes may require one carefully controlled transaction.

Request-scoped session management is useful, but it should not automatically imply that every request must commit a transaction.

Can a Unit of Work Contain Nested Transactions?

Some databases and ORMs support savepoints, which allow parts of an ongoing transaction to be rolled back independently.

SQLAlchemy exposes this through capabilities such as Session.begin_nested().

However, releasing a savepoint does not make those changes durable independently of the outer transaction. Use nested transaction behavior only when its semantics are clearly understood.

Does Unit of Work Work Across Microservices?

Unit of Work is typically used to coordinate changes within one local transactional resource.

It cannot automatically provide atomicity across independently owned databases, message brokers, and external APIs.

Cross-service business workflows often require sagas, transactional outboxes, idempotency, and reconciliation.

Should Repositories Call Commit?

When an application-level Unit of Work owns the transaction, repositories should generally not commit independently.

They may add, update, or delete persistent data within the shared session, while the Unit of Work controls commit and rollback.

For simple applications without a separate Unit of Work abstraction, a repository or service may own an individual transaction, provided that ownership is explicit and appropriate for the operation.

Conclusion

The Unit of Work Pattern coordinates related persistence changes and makes transaction boundaries explicit. It is particularly useful when a business operation requires several repositories to participate in one atomic database transaction.

Combined with the Repository Pattern, Unit of Work separates two important responsibilities: repositories provide access to domain objects, while the Unit of Work determines when the complete set of changes becomes durable.

Modern ORMs already implement many underlying Unit of Work behaviors. Application-level wrappers are therefore most valuable when they establish a meaningful architectural boundary rather than merely duplicate existing session APIs.

The pattern also has clear limits. It cannot guarantee atomic updates across independently owned services or eliminate the need for concurrency control, idempotency, and distributed failure recovery.

Key Takeaway: Coordinate changes according to the business transaction, not individual repository methods. Use one shared persistence context, commit only after all required operations succeed, roll back failed operations, and complement local transactions with appropriate distributed consistency patterns when external systems are involved.

More Articles to Read

Continue exploring related patterns and database consistency techniques on SysDesPro:

• What Is the Repository Pattern? Data Access Abstraction and Practical Examples — Understand the persistence abstraction that commonly works alongside Unit of Work.
• What Is Domain-Driven Design (DDD)? Concepts, Patterns, and Practical Examples — Learn how aggregates and consistency boundaries relate to transaction design.
• Hexagonal vs. Onion vs. Clean Architecture: Differences and When to Use Each — Compare how architectural styles isolate database and transaction implementations.

For the original pattern and ORM transaction behavior, see Martin Fowler's Unit of Work and SQLAlchemy Session Basics.