Python Database & Transaction Management — A Deep Dive

Foundations: DB-API 2.0 (PEP 249)
Connections, Cursors, and the Cost of Round-Trips
ACID and What It Actually Means at the Byte Level
Transaction Isolation Levels in Depth
Transactions as Context Managers
Savepoints and Nested Transactions
Connection Pooling
SQLAlchemy Core: Transactions the Explicit Way
SQLAlchemy ORM: Unit of Work and Session Semantics
Django ORM Transactions
Async Database Access
Locking Strategies: Optimistic vs Pessimistic
Deadlocks: Detection, Diagnosis, Avoidance
Distributed Transactions: 2PC and the Saga Pattern
Retries, Idempotency, and Exactly-Once Illusions
Testing Transactional Code
Migrations and Schema Evolution (Alembic)
Performance Considerations
Common Pitfalls Checklist

1. Foundations: DB-API 2.0 (PEP 249)

Every relational database driver in Python — sqlite3, psycopg2, psycopg3, mysqlclient, pyodbc, cx_Oracle — implements the same low-level contract defined by PEP 249. Understanding this contract is non-negotiable because every ORM (SQLAlchemy, Django, Peewee) is ultimately a code generator sitting on top of it.

Core objects

Connection: represents a session with the database. Owns the transaction state.
Cursor: the object used to execute SQL and fetch results. Cursors are cheap; connections are expensive.
Module-level attributes: apilevel, threadsafety, paramstyle (determines whether you use ?, %s, :name, etc.).

import sqlite3

conn = sqlite3.connect("app.db")   # Connection: owns transaction state
cur = conn.cursor()                # Cursor: stateless-ish, executes SQL

cur.execute("SELECT id, name FROM users WHERE active = ?", (True,))
rows = cur.fetchall()

cur.close()
conn.close()

Paramstyle matters

Different drivers use different placeholder styles:

Driver	paramstyle	Example
`sqlite3`	`qmark`	`WHERE id = ?`
`psycopg2`	`pyformat`	`WHERE id = %(id)s` or `%s`
`pyodbc`	`qmark`	`WHERE id = ?`

Never use f-strings or %-formatting to inject values into SQL — this is the #1 cause of SQL injection vulnerabilities. Always pass parameters through the driver’s placeholder mechanism so escaping is handled correctly and query plans can be cached.

# WRONG — SQL injection risk
cur.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# RIGHT — parameterized
cur.execute("SELECT * FROM users WHERE name = ?", (user_input,))

Implicit transaction start

A critical, frequently misunderstood detail: DB-API 2.0 connections start a transaction implicitly on the first statement that modifies data (and, per spec, sometimes even for SELECT, depending on the driver). There is no explicit “BEGIN” required by the API surface — but there absolutely is a transaction open. You must call commit() or rollback() explicitly; DB-API mandates connections default to autocommit = off unless the driver says otherwise (note: sqlite3 in modern Python defaults differently depending on isolation_level, and psycopg2 defaults to autocommit off).

2. Connections, Cursors, and the Cost of Round-Trips

Connections are expensive, cursors are cheap

Establishing a TCP connection, performing TLS negotiation, and authenticating against a database server can cost tens of milliseconds. This is why:

You should never open a new connection per query in a hot path.
You should open many cursors from a single connection when doing multiple sequential operations.
Production systems use connection pools (see Section 7).

Round-trip latency dominates

The single most common performance mistake in Python database code is the N+1 query problem: looping over rows and issuing a new query per row.

# BAD: N+1 queries — 1 round trip per user
for user_id in user_ids:
    cur.execute("SELECT * FROM orders WHERE user_id = ?", (user_id,))
    orders = cur.fetchall()

# GOOD: 1 round trip
cur.execute(
    "SELECT * FROM orders WHERE user_id IN ({})".format(
        ",".join("?" * len(user_ids))
    ),
    user_ids,
)
orders = cur.fetchall()

In an ORM context this manifests as forgetting to select_related / joinedload and lazily triggering a query per related object.

Server-side vs client-side cursors

Some drivers (notably psycopg2) support named (server-side) cursors that stream results from the server instead of materializing the entire result set client-side. Use these when iterating over millions of rows to avoid client memory blow-up:

with conn.cursor(name="server_side_cursor") as cur:
    cur.itersize = 2000  # fetch batch size
    cur.execute("SELECT * FROM huge_table")
    for row in cur:  # streamed, not all loaded at once
        process(row)

3. ACID and What It Actually Means at the Byte Level

Atomicity: A transaction’s writes are all-or-nothing. Implemented via write-ahead logs (WAL) or undo/redo logs. If the process crashes mid-transaction, recovery replays or discards partial writes.
Consistency: The database moves from one valid state to another — constraints (foreign keys, CHECK, UNIQUE) are enforced at commit (or immediately, depending on constraint deferrability).
Isolation: Concurrent transactions appear (to some configurable degree) as if executed serially. This is the most nuanced property and the one Python developers misunderstand most (see Section 4).
Durability: Once committed, data survives crashes — enforced via fsync to disk. Note: this is not free. Databases offer knobs (synchronous_commit = off in PostgreSQL, PRAGMA synchronous in SQLite) trading durability for throughput. Never disable these in systems where data loss on crash is unacceptable.

Why this matters in Python code

Your Python code doesn’t implement ACID — the database engine does. Your job as the application developer is to draw transaction boundaries correctly: too narrow, and you get partial/inconsistent writes; too wide, and you hold locks too long, causing contention and reduced concurrency.

# Transaction boundary too wide: locks held during slow I/O
with conn:
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = ?", (from_id,))
    send_email_confirmation(from_id)  # SLOW, unrelated I/O — locks held the whole time!
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = ?", (to_id,))

The fix: keep transactions short, move non-transactional side effects (emails, HTTP calls, logging) outside the transaction, ideally using an outbox pattern for reliability.

4. Transaction Isolation Levels in Depth

SQL defines four standard isolation levels. Python drivers expose them via connection attributes or SET TRANSACTION ISOLATION LEVEL statements.

Level	Dirty Read	Non-repeatable Read	Phantom Read	Notes
READ UNCOMMITTED	Possible	Possible	Possible	Rarely used; PostgreSQL treats it as READ COMMITTED
READ COMMITTED	No	Possible	Possible	Default in PostgreSQL, Oracle, SQL Server
REPEATABLE READ	No	No	Possible*	Default in MySQL/InnoDB; PostgreSQL’s implementation is actually snapshot isolation and prevents phantoms in practice
SERIALIZABLE	No	No	No	Implemented via Serializable Snapshot Isolation (SSI) in PostgreSQL — can abort transactions with serialization failures that the app must retry

Setting isolation level in Python

psycopg2:

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE

conn = psycopg2.connect(dsn)
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

psycopg3:

import psycopg

conn = psycopg.connect(dsn)
conn.isolation_level = psycopg.IsolationLevel.SERIALIZABLE

sqlite3 (SQLite only truly supports SERIALIZABLE at the engine level, but has its own locking modes: DEFERRED, IMMEDIATE, EXCLUSIVE):

conn = sqlite3.connect("app.db", isolation_level=None)  # autocommit mode
conn.execute("BEGIN IMMEDIATE")   # acquire write lock immediately, avoid deadlocks
...
conn.commit()

Serialization failures must be retried

Under SERIALIZABLE, the database may abort your transaction with an error like could not serialize access due to concurrent update. This is not a bug — it’s the isolation mechanism working correctly. Your code must catch this and retry:

import time
import psycopg2
from psycopg2 import errors

def run_with_serializable_retry(fn, conn, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            with conn:
                return fn(conn)
        except errors.SerializationFailure:
            conn.rollback()
            if attempt == max_attempts - 1:
                raise
            time.sleep(0.05 * (2 ** attempt))  # exponential backoff

5. Transactions as Context Managers

sqlite3 and psycopg2: connection as context manager ≠ closing the connection

A subtlety that trips up many developers: in both sqlite3 and psycopg2, using the connection object as a context manager commits or rolls back the transaction — it does not close the connection.

conn = sqlite3.connect("app.db")

with conn:  # commits on success, rolls back on exception — connection stays OPEN
    conn.execute("INSERT INTO logs(msg) VALUES (?)", ("started",))

# conn is still open here!
conn.close()  # you must close it yourself

To both manage the transaction and close the connection, nest context managers or use contextlib.closing:

from contextlib import closing

with closing(sqlite3.connect("app.db")) as conn:
    with conn:
        conn.execute("INSERT INTO logs(msg) VALUES (?)", ("done",))

Writing your own transaction context manager

For drivers/frameworks without built-in transaction context managers, write a small reusable one:

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise

with transaction(conn) as tx:
    cur = tx.cursor()
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))

6. Savepoints and Nested Transactions

True nested transactions don’t exist in most SQL engines — but savepoints provide partial rollback within a single top-level transaction. This is essential for implementing “nested” transaction semantics in application code (e.g., SQLAlchemy’s begin_nested()).

conn = psycopg2.connect(dsn)
cur = conn.cursor()

cur.execute("BEGIN")
cur.execute("INSERT INTO orders (id, status) VALUES (1, 'pending')")

cur.execute("SAVEPOINT sp1")
try:
    cur.execute("INSERT INTO order_items (order_id, sku) VALUES (1, 'BAD-SKU')")
except psycopg2.IntegrityError:
    cur.execute("ROLLBACK TO SAVEPOINT sp1")  # undo just this part
    cur.execute("RELEASE SAVEPOINT sp1")

conn.commit()  # the order row survives; the bad item insert was undone

SQLAlchemy nested transactions (savepoints)

from sqlalchemy.orm import Session

with Session(engine) as session:
    with session.begin():
        session.add(Order(id=1, status="pending"))

        try:
            with session.begin_nested():  # SAVEPOINT
                session.add(OrderItem(order_id=1, sku="BAD-SKU"))
                raise ValueError("validation failed")
        except ValueError:
            pass  # nested rollback only undoes the OrderItem insert

        # Order(id=1) is still staged for commit here

Important: each SAVEPOINT has a real cost — extra round trips and log records. Don’t use them as a substitute for proper validation; use them for genuinely optional/recoverable sub-operations.

7. Connection Pooling

Why pool?

TCP handshake + auth + (optionally) TLS negotiation for every request is unacceptable in a web application serving hundreds of requests/second. A pool keeps a set of live connections ready to be checked out and returned.

psycopg2’s built-in pools

from psycopg2 import pool

connection_pool = pool.ThreadedConnectionPool(
    minconn=2, maxconn=20, dsn="postgresql://user:pass@host/db"
)

conn = connection_pool.getconn()
try:
    with conn:
        conn.cursor().execute("SELECT 1")
finally:
    connection_pool.putconn(conn)  # ALWAYS return, even on exception

SQLAlchemy’s pooling (used even outside the ORM, via Core)

SQLAlchemy’s Engine wraps a connection pool by default (QueuePool). Key settings:

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://user:pass@host/db",
    pool_size=10,          # steady-state pool size
    max_overflow=5,        # extra connections allowed under burst load
    pool_timeout=30,       # seconds to wait for a connection before erroring
    pool_recycle=1800,     # recycle connections older than this (avoids stale TCP)
    pool_pre_ping=True,    # issue a lightweight SELECT 1 before handing out a connection
)

pool_pre_ping=True is critical in cloud environments where load balancers or firewalls silently kill idle TCP connections — without it you get cryptic OperationalError: server closed the connection unexpectedly errors.

External poolers: PgBouncer

For PostgreSQL specifically, application-level pooling (SQLAlchemy’s pool) and a network-level pooler like PgBouncer solve different problems. PgBouncer sits between your app (potentially many processes, e.g., Gunicorn workers) and Postgres, multiplexing thousands of client connections onto a small number of real server connections. Be aware: PgBouncer’s transaction pooling mode breaks features that rely on session state (SET, prepared statements, LISTEN/NOTIFY, advisory locks) unless carefully configured.

Async pools

import asyncpg

pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)

async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = $1", 1)
        await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = $1", 2)

8. SQLAlchemy Core: Transactions the Explicit Way

SQLAlchemy Core gives you SQL-level control without full ORM object mapping — good for bulk operations, reporting, or when the ORM’s identity map overhead isn’t worth it.

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://user:pass@host/db")

# engine.connect() does NOT auto-commit; you control the transaction
with engine.connect() as conn:
    with conn.begin():  # explicit transaction
        conn.execute(
            text("UPDATE accounts SET balance = balance - :amt WHERE id = :id"),
            {"amt": 100, "id": 1},
        )
        conn.execute(
            text("UPDATE accounts SET balance = balance + :amt WHERE id = :id"),
            {"amt": 100, "id": 2},
        )
    # committed here automatically if no exception, else rolled back

`engine.begin()` shortcut

with engine.begin() as conn:  # connect + begin transaction in one call
    conn.execute(text("INSERT INTO audit_log(msg) VALUES (:m)"), {"m": "transfer executed"})

Autocommit-ish behavior (2.0 style) vs legacy autocommit

SQLAlchemy 1.x had an implicit “autocommit” mode for DML outside explicit transactions — this was a frequent source of confusion and was removed in SQLAlchemy 2.0. In 2.0, every unit of work requires an explicit begin()/commit or use of Session. This is a deliberate design choice to force developers to be explicit about transaction boundaries.

9. SQLAlchemy ORM: Unit of Work and Session Semantics

The Session is not a connection — it’s a unit of work

The Session tracks object state (new, dirty, deleted) in memory and flushes changes to the database as SQL only when needed (before a query that would be affected, or on commit()). This is the Unit of Work pattern.

from sqlalchemy.orm import Session

with Session(engine) as session:
    user = session.get(User, 1)
    user.balance -= 100          # tracked as "dirty", no SQL yet

    other = session.get(User, 2) # session.get() may trigger a flush first if needed
    other.balance += 100

    session.commit()             # flush + COMMIT happens here

`session.begin()` as the outer transaction boundary

with Session(engine) as session:
    with session.begin():
        session.add(Order(...))
        session.add(OrderItem(...))
    # auto-commit on success, auto-rollback on exception

Autoflush and autocommit gotchas

Autoflush (default True) means queries executed mid-transaction will auto-flush pending changes first, so reads see your own uncommitted writes within the same session. This can cause surprise SQL at query time if you’re not tracking what’s “dirty.”
Disabling autoflush (Session(autoflush=False)) is sometimes used for performance in bulk-insert scenarios, but requires manual session.flush() calls when order matters.

Expiration after commit

By default, expire_on_commit=True — after commit(), all ORM objects are marked expired, and accessing any attribute triggers a new SELECT. This is safe but can cause subtle N+1s if you commit() inside a loop and then read attributes.

with Session(engine) as session:
    for order in session.query(Order).all():
        order.status = "shipped"
        session.commit()  # expires ALL loaded objects
        print(order.id)   # triggers a fresh SELECT just to re-fetch `id`!

Fix: batch commits outside the loop, or use expire_on_commit=False when you understand the staleness tradeoff.

Relationship loading strategies affect transaction shape

from sqlalchemy.orm import joinedload, selectinload

# N+1 risk: one query per user for orders, executed lazily on access
users = session.query(User).all()
for u in users:
    print(u.orders)  # separate SELECT per user!

# Fixed: single JOIN query
users = session.query(User).options(joinedload(User.orders)).all()

# Fixed differently: 2 queries total (1 for users, 1 IN-query for all orders)
users = session.query(User).options(selectinload(User.orders)).all()

10. Django ORM Transactions

Django wraps DB-API connections with its own transaction management layer.

`TRANSACTION` middleware / `ATOMIC_REQUESTS`

DATABASES = {
    "default": {
        ...,
        "ATOMIC_REQUESTS": True,  # wraps every view in a transaction — use with caution at scale
    }
}

This is convenient but wraps every HTTP request in a transaction, holding a connection (and potentially locks) for the request’s entire duration, including slow template rendering or external API calls. Most large Django codebases disable this globally and use explicit atomic() blocks instead.

Explicit `atomic()` blocks

from django.db import transaction

@transaction.atomic
def transfer_funds(from_id, to_id, amount):
    from_account = Account.objects.select_for_update().get(id=from_id)
    to_account = Account.objects.select_for_update().get(id=to_id)

    if from_account.balance < amount:
        raise InsufficientFunds()

    from_account.balance -= amount
    to_account.balance += amount
    from_account.save()
    to_account.save()

select_for_update() issues SELECT ... FOR UPDATE, taking a row-level lock — essential for preventing lost updates in read-modify-write patterns under concurrent access (see Section 12).

Savepoints via nested `atomic()`

with transaction.atomic():
    order = Order.objects.create(status="pending")
    try:
        with transaction.atomic():  # SAVEPOINT
            OrderItem.objects.create(order=order, sku="BAD-SKU")
            validate_sku_or_raise("BAD-SKU")
    except ValidationError:
        pass  # only the inner savepoint rolls back

`on_commit` hooks — the outbox-lite pattern

A very common Python/Django bug: sending an email or enqueuing a Celery task inside a transaction that later rolls back, causing the side effect to fire for data that never actually got persisted (or worse: the task runs before the transaction commits and reads stale/absent data due to visibility rules).

from django.db import transaction

def create_order(data):
    with transaction.atomic():
        order = Order.objects.create(**data)
        transaction.on_commit(lambda: send_confirmation_email.delay(order.id))
        # email task only enqueued if/when the transaction actually commits

Transaction isolation level configuration

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "OPTIONS": {
            "isolation_level": "serializable",  # via psycopg2 constants under the hood
        },
    }
}

11. Async Database Access

Why async matters for DB I/O

Async database drivers avoid blocking the event loop during network I/O, letting a single process handle thousands of concurrent connections — critical for high-throughput async web frameworks (FastAPI, Starlette, aiohttp).

asyncpg (PostgreSQL, very fast, binary protocol)

import asyncpg

async def transfer(pool, from_id, to_id, amount):
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.execute(
                "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                amount, from_id,
            )
            await conn.execute(
                "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
                amount, to_id,
            )

asyncpg’s transaction() context manager supports isolation level and read-only flags directly:

async with conn.transaction(isolation="serializable", readonly=False):
    ...

aiosqlite (SQLite, async wrapper — still single-writer under the hood)

import aiosqlite

async def log_event(db_path, msg):
    async with aiosqlite.connect(db_path) as conn:
        async with conn.execute("INSERT INTO logs(msg) VALUES (?)", (msg,)):
            pass
        await conn.commit()

Important nuance: SQLite itself only allows one writer at a time regardless of async wrapping — aiosqlite doesn’t change SQLite’s fundamental single-writer concurrency model, it just avoids blocking the Python event loop while waiting on the (synchronous, threaded) SQLite driver.

SQLAlchemy 2.0 async ORM

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")

async def transfer(from_id, to_id, amount):
    async with AsyncSession(engine) as session:
        async with session.begin():
            from_acc = await session.get(Account, from_id)
            to_acc = await session.get(Account, to_id)
            from_acc.balance -= amount
            to_acc.balance += amount
        # commits automatically on successful exit

Mixing sync and async: the #1 async DB mistake

Calling a synchronous driver (e.g., psycopg2, plain sqlite3) directly inside an async def blocks the entire event loop, stalling every other coroutine. If you must use a sync driver in an async app, offload it:

import asyncio

async def run_sync_query(conn, query, params):
    return await asyncio.to_thread(conn.execute, query, params)

Better: use the native async driver (asyncpg, asyncmy for MySQL) instead of wrapping a sync one.

12. Locking Strategies: Optimistic vs Pessimistic

Pessimistic locking: `SELECT ... FOR UPDATE`

Assumes conflicts are likely; acquires a row lock upfront, blocking other transactions from modifying (or, with FOR UPDATE, sometimes even reading) the same rows until commit/rollback.

# psycopg2 / raw SQL
cur.execute("BEGIN")
cur.execute("SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (account_id,))
balance = cur.fetchone()[0]
cur.execute("UPDATE accounts SET balance = %s WHERE id = %s", (balance - 100, account_id))
cur.execute("COMMIT")

# SQLAlchemy ORM
account = session.query(Account).filter_by(id=account_id).with_for_update().one()
account.balance -= 100
session.commit()

FOR UPDATE SKIP LOCKED is a powerful variant for building job queues directly in SQL — workers grab the next available row without blocking on rows other workers already hold:

cur.execute("""
    SELECT id FROM job_queue
    WHERE status = 'pending'
    ORDER BY created_at
    FOR UPDATE SKIP LOCKED
    LIMIT 1
""")

Optimistic locking: version columns

Assumes conflicts are rare; instead of locking, checks a version number (or timestamp) at write time and fails the write if it changed since read.

# schema: accounts(id, balance, version)

cur.execute("SELECT balance, version FROM accounts WHERE id = %s", (account_id,))
balance, version = cur.fetchone()

new_balance = balance - 100
cur.execute(
    "UPDATE accounts SET balance = %s, version = version + 1 "
    "WHERE id = %s AND version = %s",
    (new_balance, account_id, version),
)

if cur.rowcount == 0:
    raise ConcurrentModificationError("row was modified by another transaction")

SQLAlchemy has built-in support via version_id_col:

class Account(Base):
    __tablename__ = "accounts"
    id = Column(Integer, primary_key=True)
    balance = Column(Numeric)
    version_id = Column(Integer, nullable=False)

    __mapper_args__ = {"version_id_col": version_id}

try:
    session.commit()
except StaleDataError:
    session.rollback()
    # retry: reload and reapply the change

When to use which

Scenario	Prefer
High contention (many writers on same rows)	Pessimistic
Low contention, high read/write ratio	Optimistic
Long user “think time” between read and write (e.g., editing a form)	Optimistic (never hold a DB lock across user think-time!)
Job queues / task dispatch	Pessimistic with `SKIP LOCKED`

13. Deadlocks: Detection, Diagnosis, Avoidance

A deadlock occurs when Transaction A holds a lock Transaction B needs, and vice versa. Databases detect this cycle and abort one transaction (the “victim”) with an error — this is not a hang, it’s an exception you must handle.

Reproducing the classic deadlock

# Transaction 1                      # Transaction 2
UPDATE accounts SET ... WHERE id=1;  UPDATE accounts SET ... WHERE id=2;
UPDATE accounts SET ... WHERE id=2;  UPDATE accounts SET ... WHERE id=1;
# T1 waits for T2's lock on id=2     # T2 waits for T1's lock on id=1 → DEADLOCK

Catching deadlocks in Python

import psycopg2
from psycopg2 import errors
import time

def transfer_with_deadlock_retry(conn, from_id, to_id, amount, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            with conn:
                cur = conn.cursor()
                cur.execute(
                    "UPDATE accounts SET balance = balance - %s WHERE id = %s",
                    (amount, from_id),
                )
                cur.execute(
                    "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                    (amount, to_id),
                )
            return
        except errors.DeadlockDetected:
            if attempt == max_attempts - 1:
                raise
            time.sleep(0.01 * (attempt + 1))

Prevention: consistent lock ordering

The single most effective deadlock-avoidance technique: always acquire locks in the same order across all code paths, e.g., always by ascending primary key.

def transfer(conn, id_a, id_b, amount):
    first, second = sorted([id_a, id_b])  # canonical ordering
    ...
    # lock/update `first` before `second`, always

14. Distributed Transactions: 2PC and the Saga Pattern

Two-Phase Commit (2PC)

DB-API 2.0 exposes tpc_begin(), tpc_prepare(), tpc_commit(), tpc_rollback() for drivers that support XA-style distributed transactions (e.g., psycopg2 against PostgreSQL’s PREPARE TRANSACTION).

conn = psycopg2.connect(dsn)
xid = conn.xid(42, "transfer", "branch-1")

conn.tpc_begin(xid)
cur = conn.cursor()
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")

conn.tpc_prepare()   # phase 1: all participants prepare
# --- coordinator confirms ALL participants prepared successfully ---
conn.tpc_commit()    # phase 2: commit
# or conn.tpc_rollback() if any participant failed to prepare

In practice, 2PC is rarely used in modern Python microservice architectures because:

It requires a transaction coordinator and long-held locks across services, hurting availability.
Most modern databases require enabling it explicitly (max_prepared_transactions in PostgreSQL), and orphaned prepared transactions can block vacuum/cleanup if the coordinator crashes.

Saga pattern: the pragmatic alternative

Instead of atomicity across services, a Saga is a sequence of local transactions, each with a corresponding compensating transaction to undo it if a later step fails.

class TransferSaga:
    def execute(self, from_account, to_account, amount):
        steps_completed = []
        try:
            self.debit(from_account, amount)
            steps_completed.append(("credit", from_account, amount))

            self.credit(to_account, amount)
            steps_completed.append(("debit", to_account, amount))

            self.notify(from_account, to_account, amount)
        except Exception:
            self._compensate(steps_completed)
            raise

    def _compensate(self, steps_completed):
        for action, account, amount in reversed(steps_completed):
            getattr(self, action)(account, amount)  # e.g. re-credit or re-debit

Combine sagas with an outbox table (write the “event to publish” in the same local transaction as the business change, then a separate poller publishes it) to guarantee at-least-once delivery of cross-service events without 2PC.

def create_order(session, order_data):
    with session.begin():
        order = Order(**order_data)
        session.add(order)
        session.add(OutboxEvent(
            event_type="OrderCreated",
            payload=json.dumps(order_data),
        ))
    # a separate background worker polls OutboxEvent and publishes to a message broker

15. Retries, Idempotency, and Exactly-Once Illusions

There is no such thing as exactly-once delivery over a network

If your client sends a write and the connection drops before the response arrives, you cannot know whether the write succeeded or not. The only sound engineering response is idempotency: design operations so that retrying them is safe.

Idempotency keys

def process_payment(session, idempotency_key, amount, account_id):
    existing = session.query(PaymentAttempt).filter_by(
        idempotency_key=idempotency_key
    ).one_or_none()

    if existing is not None:
        return existing.result  # already processed — return the same result

    with session.begin():
        result = charge_account(account_id, amount)
        session.add(PaymentAttempt(
            idempotency_key=idempotency_key,
            result=result,
        ))
    return result

The uniqueness constraint on idempotency_key should be enforced at the database level (UNIQUE index), not just checked-then-inserted in application code (which itself has a race condition — use INSERT ... ON CONFLICT DO NOTHING or catch the IntegrityError).

try:
    session.add(PaymentAttempt(idempotency_key=key, ...))
    session.commit()
except IntegrityError:
    session.rollback()
    # another concurrent request already inserted this key — fetch and return its result

Retry decorators with backoff (using `tenacity`)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from psycopg2 import errors

@retry(
    retry=retry_if_exception_type((errors.SerializationFailure, errors.DeadlockDetected)),
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=0.05, max=2),
)
def run_transaction(fn, conn):
    with conn:
        return fn(conn)

Never blindly retry non-idempotent operations (e.g., “insert a new row with an auto-generated ID”) without an idempotency key — you will create duplicates.

16. Testing Transactional Code

The “wrap each test in a rolled-back transaction” pattern

The standard approach for fast, isolated database tests: begin a transaction (and, in SQLAlchemy, a nested savepoint) before each test, and roll it back afterward — no data ever actually persists, and tests don’t interfere with each other.

import pytest
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    session.begin_nested()  # SAVEPOINT so session.commit() inside test code doesn't escape

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(sess, trans):
        if trans.nested and not trans._parent.nested:
            sess.begin_nested()

    yield session

    session.close()
    transaction.rollback()  # undo EVERYTHING, including any "committed" data
    connection.close()

Django’s `TestCase` does this automatically

from django.test import TestCase

class TransferTests(TestCase):
    # Django wraps each test method in a transaction and rolls it back automatically
    def test_transfer_moves_funds(self):
        transfer_funds(self.acc1.id, self.acc2.id, 100)
        self.acc1.refresh_from_db()
        self.assertEqual(self.acc1.balance, 900)

Note: TestCase cannot test code that itself calls transaction.atomic() and expects real commit/rollback boundaries to matter (e.g., testing on_commit hooks) — use TransactionTestCase for that, at the cost of much slower tests (real commits + table truncation between tests).

from django.test import TransactionTestCase

class OnCommitHookTests(TransactionTestCase):
    def test_email_sent_only_after_commit(self):
        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            create_order(data)
        self.assertEqual(len(callbacks), 1)

Testing deadlocks and race conditions

Concurrency bugs need real concurrency to surface. Use threads or multiprocessing against a real (often containerized, via testcontainers) database:

import threading

def test_concurrent_updates_dont_lose_writes(pg_container):
    errors = []
    def worker():
        try:
            transfer_funds(acc1, acc2, 10)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=worker) for _ in range(20)]
    for t in threads: t.start()
    for t in threads: t.join()

    # assert final balance is correct regardless of interleaving

17. Migrations and Schema Evolution (Alembic)

Autogenerate vs hand-written migrations

# alembic revision --autogenerate -m "add version column to accounts"

Autogenerate compares your ORM models to the live schema and drafts a migration — always review the generated script; autogenerate frequently misses data migrations, index renames, and check-constraint changes.

def upgrade():
    op.add_column("accounts", sa.Column("version_id", sa.Integer(), nullable=True))
    op.execute("UPDATE accounts SET version_id = 0")   # backfill existing rows
    op.alter_column("accounts", "version_id", nullable=False)

def downgrade():
    op.drop_column("accounts", "version_id")

Zero-downtime migrations require multi-step deploys

Adding a NOT NULL column safely in a live system with zero downtime typically requires three separate deploys:

Add the column as nullable with a default, deploy code that writes to it.
Backfill existing rows (in batches, to avoid long locks), deploy code that reads it.
Alter the column to NOT NULL once backfill is confirmed complete.

# Step 1
op.add_column("accounts", sa.Column("version_id", sa.Integer(), server_default="0"))

# Step 2 (separate migration, run in batches to avoid locking the whole table)
op.execute("""
    UPDATE accounts SET version_id = 0
    WHERE version_id IS NULL
    AND id IN (SELECT id FROM accounts WHERE version_id IS NULL LIMIT 10000)
""")  # repeat until 0 rows affected

# Step 3 (separate migration, after backfill confirmed complete)
op.alter_column("accounts", "version_id", nullable=False)

Adding an index on a large PostgreSQL table should use CREATE INDEX CONCURRENTLY to avoid locking writes — but this cannot run inside a transaction, so Alembic migrations doing this must disable transactional DDL for that revision:

def upgrade():
    with op.get_context().autocommit_block():
        op.create_index(
            "ix_accounts_email", "accounts", ["email"],
            postgresql_concurrently=True,
        )

18. Performance Considerations

Batch writes: use executemany() or INSERT ... VALUES (...), (...), (...) multi-row inserts instead of one INSERT per row.
copy_from / COPY: for bulk-loading into PostgreSQL, psycopg2’s copy_expert is orders of magnitude faster than row-by-row inserts.
Prepared statements: reused parameterized queries let the database cache query plans — most drivers do this transparently, but be aware some connection poolers (PgBouncer transaction mode) disable prepared statement reuse across pooled connections.
Don’t hold transactions open across non-DB I/O (HTTP calls, disk writes, time.sleep) — this is the single biggest cause of lock contention in production incidents.
Bulk operations bypass the ORM’s per-object overhead: SQLAlchemy’s bulk_insert_mappings / Core insert() executemany avoid instantiating full ORM objects.
Measure with EXPLAIN ANALYZE, not intuition — index usage assumptions are frequently wrong, especially after data grows.

Table of Contents