Introduction

Every serious backend developer eventually faces the same problem: you need to make multiple changes to a database as part of a single business operation, and you need all of them to succeed or none of them to go through. Partial updates are worse than no updates at all - they leave your data in an inconsistent state that can be nearly impossible to debug in production.
This is not a new problem. Enterprise developers have been solving it for decades, and Martin Fowler documented the canonical solution in his 2002 book Patterns of Enterprise Application Architecture: the Unit of Work pattern.
In this article we are going to go deep on what Unit of Work is, why it exists, how it works internally, and how to build a clean, production-quality implementation from scratch in Python using only the standard library. By the end you will have a working implementation you can adapt to any project, and a solid understanding of how popular frameworks like SQLAlchemy and Django ORM implement this pattern under the hood.
The full source code is available on GitHub:
👉 github.com/diegocastillo12/unit-of-work-python

• - ## Background: What is the Unit of Work Pattern? The Unit of Work pattern is part of Martin Fowler's catalog of Patterns of Enterprise Application Architecture (PoEAA), a collection of battle-tested solutions for common problems in enterprise software design. Fowler defines it as follows: > "A Unit of Work maintains a list of objects affected by a business transaction and coordinates the writing out of changes and the resolution of concurrency problems." Let's unpack that definition carefully. "Maintains a list of objects affected by a business transaction" - this means the Unit of Work acts as a tracker. When your business logic creates a new object, modifies an existing one, or marks one for deletion, it does not immediately write to the database. Instead, it registers the change with the Unit of Work, which keeps an in-memory list of everything that needs to happen. "Coordinates the writing out of changes" - when the business transaction is complete and everything looks correct, the Unit of Work flushes all the accumulated changes to the database in a single coordinated operation. This is the commit() call. "Resolution of concurrency problems" - by batching all writes into a single transaction, the Unit of Work also helps manage concurrent access. Either all changes go through together or none do, which prevents other processes from seeing partial states. The pattern is closely related to two others from the same catalog: the Identity Map (which ensures that each database record is loaded only once per transaction, preventing duplicate in-memory objects) and the Repository (which abstracts the data access layer and uses the Unit of Work to coordinate persistence). Together these three patterns form the backbone of most modern ORM implementations.
• - ## The Problem it Solves: A Concrete Example To understand why Unit of Work matters, let's look at what happens without it. Imagine you are building an e-commerce platform. When a customer places an order, your system needs to:
• Create a new order record in the orders table
• Deduct the purchased quantity from the inventory table
• Record the total price for billing Without Unit of Work, a naive implementation might look like this:

def process_order_naive(conn, item_id, quantity):
 cursor = conn.cursor()
# Step 1: insert the order
 cursor.execute("INSERT INTO orders (item_id, quantity) VALUES (?, ?)", (item_id, quantity))
 conn.commit() # committed immediately
# Step 2: update inventory
 cursor.execute("UPDATE inventory SET stock = stock - ? WHERE id = ?", (quantity, item_id))
 conn.commit() # committed immediately
# Step 3: record billing
 cursor.execute("INSERT INTO billing (item_id, quantity) VALUES (?, ?)", (item_id, quantity))
 conn.commit() # committed immediately

This looks reasonable but has a critical flaw. If step 2 fails - say, due to a network error, a constraint violation, or a bug - the order already exists in the database but the stock was never decremented. You now have:

• An order that references a product
• Inventory that still shows the original stock level
• No billing record Your database is now inconsistent. A customer placed an order and your system thinks the stock is still available. The next customer might be able to order the same item even though it is already sold. Debugging this in production at 2am is not a good time. The root cause is simple: each operation was committed independently. There was no coordination. With Unit of Work, all three operations are registered in memory and only committed together as a single atomic transaction. If any step fails, all three are rolled back automatically. The database either reflects the complete order or it reflects nothing - there is no in-between state.
• - ## Core Concepts in Detail Before diving into code, let's understand the four core operations that define a Unit of Work: ### register_new(obj) Called when a new domain object is created that does not yet exist in the database. The Unit of Work adds it to an internal new_objects list. On commit, these become INSERT statements. ### register_dirty(obj) Called when an existing domain object has been modified. "Dirty" is a classic database term meaning "modified but not yet persisted". The Unit of Work adds it to a dirty_objects list. On commit, these become UPDATE statements. ### register_deleted(obj) Called when a domain object needs to be removed from the database. The Unit of Work adds it to a deleted_objects list. On commit, these become DELETE statements. ### commit() The moment of truth. When called, the Unit of Work iterates over all three lists and generates the corresponding SQL statements, executes them inside a single database transaction, and calls connection.commit(). If anything fails at any point, it calls connection.rollback() and raises the exception so the caller knows something went wrong. The beauty of this design is that your business logic never writes to the database directly. It only works with in-memory Python objects and registers intent. The Unit of Work handles all the SQL coordination.
• - ## Implementation: Step by Step Now let's build the full implementation. We are using Python's built-in sqlite3 library and dataclasses - no external dependencies required. ### Step 1: Define the Domain Models

import sqlite3
from dataclasses import dataclass
from typing import List, Any, Optional
@dataclass
class InventoryItem:
 id: int
 name: str
 price: float
 stock: int
@dataclass
class Order:
 id: Optional[int]
 item_id: int
 quantity: int
 total_price: float

We use Python dataclasses for simplicity. In a real project these would be your domain entities - classes that represent core business concepts in your system, completely ignorant of how they are persisted.
Notice that Order.id is Optional[int] - when we create a new order it does not have a database ID yet. The database will assign one on insert.

Step 2: Build the Unit of Work Class

class UnitOfWork:
 """
 Implements the Unit of Work design pattern.
 Maintains a list of objects affected by a business transaction
 and coordinates the writing out of changes and the resolution
 of concurrency problems.
Reference: Fowler, M. (2002). Patterns of Enterprise Application
 Architecture. Addison-Wesley.
 https://martinfowler.com/eaaCatalog/unitOfWork.html
 """
def __init__(self, connection: sqlite3.Connection):
 self.connection = connection
 self.new_objects: List[Any] = []
 self.dirty_objects: List[Any] = []
 self.deleted_objects: List[Any] = []
def __enter__(self):
 return self
def __exit__(self, exc_type, exc_val, exc_tb):
 if exc_type is not None:
 self.rollback()
 else:
 self.commit()
 self._clear()
def _clear(self):
 self.new_objects.clear()
 self.dirty_objects.clear()
 self.deleted_objects.clear()
def register_new(self, obj: Any):
 if obj not in self.new_objects:
 self.new_objects.append(obj)
def register_dirty(self, obj: Any):
 if obj not in self.dirty_objects:
 self.dirty_objects.append(obj)
def register_deleted(self, obj: Any):
 if obj not in self.deleted_objects:
 self.deleted_objects.append(obj)
def commit(self):
 cursor = self.connection.cursor()
 try:
 # Process new objects - INSERT
 for obj in self.new_objects:
 if isinstance(obj, Order):
 cursor.execute(
 "INSERT INTO orders (item_id, quantity, total_price) "
 "VALUES (?, ?, ?)",
 (obj.item_id, obj.quantity, obj.total_price)
 )
 elif isinstance(obj, InventoryItem):
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) "
 "VALUES (?, ?, ?, ?)",
 (obj.id, obj.name, obj.price, obj.stock)
 )
# Process dirty objects - UPDATE
 for obj in self.dirty_objects:
 if isinstance(obj, InventoryItem):
 cursor.execute(
 "UPDATE inventory SET name=?, price=?, stock=? WHERE id=?",
 (obj.name, obj.price, obj.stock, obj.id)
 )
 elif isinstance(obj, Order):
 cursor.execute(
 "UPDATE orders SET item_id=?, quantity=?, total_price=? WHERE id=?",
 (obj.item_id, obj.quantity, obj.total_price, obj.id)
 )
# Process deleted objects - DELETE
 for obj in self.deleted_objects:
 if isinstance(obj, InventoryItem):
 cursor.execute(
 "DELETE FROM inventory WHERE id=?", (obj.id,)
 )
 elif isinstance(obj, Order):
 cursor.execute(
 "DELETE FROM orders WHERE id=?", (obj.id,)
 )
self.connection.commit()
 print("[UoW] Transaction committed successfully.")
except Exception as e:
 self.rollback()
 raise e
def rollback(self):
 self.connection.rollback()
 print("[UoW] Transaction rolled back.")

A few things worth highlighting here:
Context manager support - implementing __enter__ and __exit__ allows the with statement. If an exception is raised anywhere inside the with block, Python automatically calls __exit__ with the exception information, which triggers rollback(). If everything succeeds, it calls commit(). This means the caller does not need any try/except logic - the Unit of Work handles it.
Duplicate guard - both register_new and register_dirty check if obj not in list before appending. This prevents the same object from being processed twice if your business logic accidentally registers it more than once.
Clear on exit - _clear() is called in __exit__ regardless of success or failure. This ensures the Unit of Work is clean and ready for reuse after the transaction completes.

Step 3: Set Up the Database

def setup_db() -> sqlite3.Connection:
 """
 Initializes an in-memory SQLite database with schema and seed data.
 In a real application this would point to a persistent database.
 """
 conn = sqlite3.connect(":memory:")
 cursor = conn.cursor()
cursor.execute("""
 CREATE TABLE inventory (
 id INTEGER PRIMARY KEY,
 name TEXT NOT NULL,
 price REAL NOT NULL,
 stock INTEGER NOT NULL
 )
 """)
cursor.execute("""
 CREATE TABLE orders (
 id INTEGER PRIMARY KEY AUTOINCREMENT,
 item_id INTEGER NOT NULL,
 quantity INTEGER NOT NULL,
 total_price REAL NOT NULL,
 FOREIGN KEY (item_id) REFERENCES inventory(id)
 )
 """)
# Seed with two products
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) VALUES (1, 'Laptop', 1200.0, 10)"
 )
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) VALUES (2, 'Smartphone', 800.0, 5)"
 )
conn.commit()
 return conn

We use :memory: for simplicity - this creates a temporary SQLite database that lives only in RAM. For a real application you would replace this with a connection string to your actual database.

Step 4: Business Logic Using Unit of Work

def process_order(conn: sqlite3.Connection, item_id: int, quantity: int):
 """
 Processes a purchase order using the Unit of Work pattern.
 All database changes are coordinated through the UoW - none
 of them touch the database directly until commit() is called.
 """
 cursor = conn.cursor()
 cursor.execute(
 "SELECT id, name, price, stock FROM inventory WHERE id=?", (item_id,)
 )
 row = cursor.fetchone()
if not row:
 raise ValueError(f"Product with ID {item_id} not found.")
# Load the domain object into memory
 item = InventoryItem(id=row[0], name=row[1], price=row[2], stock=row[3])
print(
 f"\n[Process] Attempting to purchase {quantity}x "
 f"'{item.name}' (Available stock: {item.stock})"
 )
# The entire business transaction runs inside the UoW context
 with UnitOfWork(conn) as uow:
# Validate business rules BEFORE registering any changes
 if item.stock < quantity:
 raise ValueError(
 f"Insufficient stock for '{item.name}'. "
 f"Requested: {quantity}, Available: {item.stock}"
 )
# Modify the in-memory object and register it as dirty
 item.stock -= quantity
 uow.register_dirty(item)
# Create the new order object and register it as new
 total_price = item.price * quantity
 order = Order(
 id=None,
 item_id=item_id,
 quantity=quantity,
 total_price=total_price
 )
 uow.register_new(order)
# At this point nothing has been written to the database yet.
 # When the 'with' block exits successfully, __exit__ calls commit()
 # and both changes (stock update + order insert) go through together.

This is the key design point. Inside the with block, we:

1. Load the current state from the database
2. Validate the business rule (enough stock available)
3. Modify the in-memory objects
4. Register the changes with the Unit of Work Nothing touches the database until the with block exits cleanly. If the stock check fails and raises a ValueError, the __exit__ method sees the exception and calls rollback() instead of commit(). The database remains untouched. ### Step 5: Utility and Main Execution

def print_db_state(conn: sqlite3.Connection):
 cursor = conn.cursor()
 print("\n - - DATABASE CURRENT STATE - -")
 print("Inventory Table:")
 cursor.execute("SELECT * FROM inventory")
 for row in cursor.fetchall():
 print(f" ID: {row[0]} | Name: {row[1]:<12} | Price: ${row[2]:.2f} | Stock: {row[3]}")
print("Orders Table:")
 cursor.execute("SELECT * FROM orders")
 orders = cursor.fetchall()
 if not orders:
 print(" (empty)")
 for row in orders:
 print(f" ID: {row[0]} | Item ID: {row[1]} | Quantity: {row[2]} | Total: ${row[3]:.2f}")
 print(" - - - - - - - - - - - - - - - ")
if __name__ == "__main__":
 conn = setup_db()
print("=== Initial Database State ===")
 print_db_state(conn)
print("\n=== Test Case 1: Successful Order (3 Laptops) ===")
 try:
 process_order(conn, item_id=1, quantity=3)
 print("Result: Order processed successfully.")
 except Exception as e:
 print(f"Result: Failed - {e}")
 print_db_state(conn)
print("\n=== Test Case 2: Failed Order (6 Smartphones, only 5 in stock) ===")
 try:
 process_order(conn, item_id=2, quantity=6)
 print("Result: Order processed successfully.")
 except Exception as e:
 print(f"Result: Failed as expected - {e}")
 print_db_state(conn)

• - ## Expected Output Running python unit_of_work.py produces: