Postgres Code Review

PostgreSQL Code Review

Quick Reference

Review Checklist

• [ ] WHERE/JOIN columns have appropriate indexes
• [ ] Composite indexes match query patterns (column order matters)
• [ ] JSONB columns use GIN indexes when queried
• [ ] Using proper JSONB operators (->, ->>, @>, ?)
• [ ] Connection pool configured with appropriate limits
• [ ] Connections properly released (context managers, try/finally)
• [ ] Appropriate transaction isolation level for use case
• [ ] No long-running transactions holding locks
• [ ] Advisory locks used for application-level coordination
• [ ] Queries use parameterized statements (no SQL injection)

Gates (before reporting findings)

Use this sequence so conclusions stay evidence-bound (not “I checked mentally”):

1. Scope — Record the concrete paths (and line ranges or symbols if helpful) for the SQL, DDL/migrations, and connection code under review. Pass: every subsystem you critique (queries, JSONB, pool, transactions) has at least one cited path. 2. SQL/DDL citation for performance claims — Index, sequential-scan, JSONB-operator, and plan-related findings must point to the exact statement or schema (quoted excerpt or file:line). Pass: each such finding includes that citation. 3. Binding check before injection flags — Only assert SQL-injection risk after locating how SQL and values are combined (bound parameters vs string concat/format/f-strings). Pass: you name the mechanism you saw in code for each flagged callsite.

Then load the relevant reference doc from Quick Reference and walk the Review Checklist.

When to Load References

• Reviewing SELECT queries with WHERE/JOIN → indexes.md
• Reviewing JSONB columns or JSON operations → jsonb.md
• Reviewing database connection code → connections.md
• Reviewing BEGIN/COMMIT or concurrent updates → transactions.md

Review Questions

1. Will this query use an index or perform a sequential scan? 2. Are JSONB operations using appropriate operators and indexes? 3. Are database connections properly managed and released? 4. Is the transaction isolation level appropriate for this operation? 5. Could this cause deadlocks or long-running locks?

Connections

Critical Anti-Patterns

1. Not Using Connection Pooling

Problem: Creating new connection per request is slow and exhausts database connections.

# BAD: New connection every time
def get_user(user_id: int):
    conn = psycopg2.connect(
        host='localhost',
        database='mydb',
        user='user',
        password='password'
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    result = cursor.fetchone()
    conn.close()
    return result

# GOOD: Use connection pool
from psycopg2.pool import ThreadedConnectionPool

pool = ThreadedConnectionPool(
    minconn=5,
    maxconn=20,
    host='localhost',
    database='mydb',
    user='user',
    password='password'
)

def get_user(user_id: int):
    conn = pool.getconn()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cursor.fetchone()
    finally:
        pool.putconn(conn)

2. Connection Leaks

Problem: Not releasing connections back to pool causes starvation.

# BAD: Connection leaked on error
def get_user(user_id: int):
    conn = pool.getconn()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    result = cursor.fetchone()
    pool.putconn(conn)  # Not called if error occurs!
    return result

# GOOD: Always release in finally block
def get_user(user_id: int):
    conn = pool.getconn()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cursor.fetchone()
    finally:
        pool.putconn(conn)

# BETTER: Use context manager
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = pool.getconn()
    try:
        yield conn
    finally:
        pool.putconn(conn)

def get_user(user_id: int):
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cursor.fetchone()

3. Wrong Pool Size

Problem: Pool too small (connection starvation) or too large (resource waste).

# BAD: Pool size not based on workload
pool = ThreadedConnectionPool(minconn=1, maxconn=100)

# GOOD: Size based on concurrent requests and database limits
# Rule of thumb: (num_cores * 2) + effective_spindle_count
# For web server: match number of worker threads/processes
pool = ThreadedConnectionPool(
    minconn=5,   # Keep warm connections
    maxconn=20,  # Max concurrent requests
    host='localhost',
    database='mydb'
)

# Check PostgreSQL connection limit
# SHOW max_connections;  -- Default 100
# Ensure pool max < max_connections across all app instances

4. No Connection Timeout

Problem: Application hangs waiting for connections.

# BAD: No timeout, hangs indefinitely
pool = ThreadedConnectionPool(minconn=5, maxconn=20)
conn = pool.getconn()  # Blocks forever if pool exhausted

# GOOD: Use timeout
pool = ThreadedConnectionPool(minconn=5, maxconn=20)
conn = pool.getconn(timeout=5)  # Raises error after 5 seconds
if conn is None:
    raise Exception("Could not get database connection")

# BETTER: Use asyncpg with async/await
import asyncpg

pool = await asyncpg.create_pool(
    host='localhost',
    database='mydb',
    min_size=5,
    max_size=20,
    timeout=5,
    command_timeout=30  # Query timeout
)

async def get_user(user_id: int):
    async with pool.acquire() as conn:
        return await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1",
            user_id
        )

5. Not Setting Statement Timeout

Problem: Long-running queries hold connections and locks.

# BAD: No query timeout
async def expensive_query():
    async with pool.acquire() as conn:
        # Could run for hours, holding connection
        return await conn.fetch("SELECT * FROM huge_table")

# GOOD: Set statement timeout
async def expensive_query():
    async with pool.acquire() as conn:
        await conn.execute("SET statement_timeout = '30s'")
        try:
            return await conn.fetch("SELECT * FROM huge_table")
        except asyncpg.QueryCanceledError:
            raise TimeoutError("Query took too long")

# BETTER: Set at connection level
pool = await asyncpg.create_pool(
    host='localhost',
    database='mydb',
    command_timeout=30,  # 30 second timeout for all queries
    server_settings={'statement_timeout': '30000'}  # milliseconds
)

6. Not Using PgBouncer

Problem: Application connection pool doesn't reduce database connections.

# BAD: Each app instance has its own pool
# 3 app servers * 20 connections = 60 database connections

# GOOD: Use PgBouncer for connection pooling
# pgbouncer.ini
[databases]
mydb = host=localhost port=5432 dbname=mydb

[pgbouncer]
listen_addr = *
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction  # or session
max_client_conn = 1000   # Application connections
default_pool_size = 20   # Database connections
reserve_pool_size = 5

# Application connects to PgBouncer instead of PostgreSQL
pool = await asyncpg.create_pool(
    host='localhost',
    port=6432,  # PgBouncer port
    database='mydb'
)
# Now 3 app servers * 20 connections = 60 app connections
# But only 20 database connections via PgBouncer

7. Holding Connections During I/O

Problem: Holding database connection while doing network/file I/O.

# BAD: Holding connection during API call
async def process_user(user_id: int):
    async with pool.acquire() as conn:
        user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

        # Holding connection during external API call!
        async with httpx.AsyncClient() as client:
            response = await client.get(f"https://api.example.com/users/{user_id}")

        await conn.execute(
            "UPDATE users SET api_data = $1 WHERE id = $2",
            response.json(), user_id
        )

# GOOD: Release connection during I/O
async def process_user(user_id: int):
    async with pool.acquire() as conn:
        user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

    # Connection released during API call
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")

    async with pool.acquire() as conn:
        await conn.execute(
            "UPDATE users SET api_data = $1 WHERE id = $2",
            response.json(), user_id
        )

8. Not Monitoring Connection Pool

Problem: Can't diagnose connection starvation or leaks.

# BAD: No visibility into pool state
pool = ThreadedConnectionPool(minconn=5, maxconn=20)

# GOOD: Monitor pool metrics
import logging

logger = logging.getLogger(__name__)

@contextmanager
def get_db_connection():
    logger.info(f"Pool: {pool._used}/{pool._maxconn} connections used")
    conn = pool.getconn()
    try:
        yield conn
    finally:
        pool.putconn(conn)

# BETTER: Use metrics library
from prometheus_client import Gauge

db_connections_used = Gauge('db_connections_used', 'Database connections in use')
db_connections_max = Gauge('db_connections_max', 'Max database connections')

@contextmanager
def get_db_connection():
    conn = pool.getconn()
    db_connections_used.inc()
    try:
        yield conn
    finally:
        pool.putconn(conn)
        db_connections_used.dec()

db_connections_max.set(pool._maxconn)

Review Questions

1. Is connection pooling used? 2. Are connections always released (try/finally or context manager)? 3. Is pool size appropriate for workload? 4. Are connection and statement timeouts configured? 5. Would PgBouncer help reduce database connections? 6. Are connections released during I/O operations? 7. Is connection pool health monitored? 8. Are connection errors handled and logged?

Indexes

Critical Anti-Patterns

1. Missing Index on WHERE Clause

Problem: Sequential scan on large tables causes slow queries.

-- BAD: No index on email
SELECT * FROM users WHERE email = 'user@example.com';

-- GOOD: Create index
CREATE INDEX idx_users_email ON users(email);
SELECT * FROM users WHERE email = 'user@example.com';

# Check query plan
EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'user@example.com';
# Look for "Seq Scan" (bad) vs "Index Scan" (good)

2. Wrong Column Order in Composite Index

Problem: Index not used if query doesn't match leftmost columns.

-- BAD: Index doesn't match query pattern
CREATE INDEX idx_orders_wrong ON orders(status, user_id);
SELECT * FROM orders WHERE user_id = 123;  -- Won't use index!

-- GOOD: Match query pattern
CREATE INDEX idx_orders_user_status ON orders(user_id, status);
SELECT * FROM orders WHERE user_id = 123;  -- Uses index
SELECT * FROM orders WHERE user_id = 123 AND status = 'pending';  -- Uses index

Rule: Put high-selectivity columns first, match WHERE clause order.

3. Not Using Partial Indexes

Problem: Indexing entire table when only subset is queried.

-- BAD: Index includes all rows
CREATE INDEX idx_orders_status ON orders(status);

-- GOOD: Only index active orders
CREATE INDEX idx_orders_active ON orders(user_id, created_at)
WHERE status = 'active';

SELECT * FROM orders
WHERE status = 'active' AND user_id = 123
ORDER BY created_at DESC;  -- Uses partial index

4. Missing Index on Foreign Keys

Problem: Slow JOINs and cascading deletes.

-- BAD: No index on foreign key
CREATE TABLE order_items (
    id SERIAL PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    product_id INTEGER REFERENCES products(id)
);

-- GOOD: Index foreign keys
CREATE TABLE order_items (
    id SERIAL PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    product_id INTEGER REFERENCES products(id)
);
CREATE INDEX idx_order_items_order_id ON order_items(order_id);
CREATE INDEX idx_order_items_product_id ON order_items(product_id);

5. Not Using EXPLAIN ANALYZE

Problem: Guessing instead of measuring query performance.

# BAD: Assuming query is fast
cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))

# GOOD: Verify with EXPLAIN
cursor.execute("""
    EXPLAIN ANALYZE
    SELECT * FROM orders WHERE user_id = %s
""", (user_id,))
print(cursor.fetchall())
# Check: Index Scan vs Seq Scan, actual time, rows

# Then run actual query
cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))

6. Over-Indexing

Problem: Slows down writes, wastes space.

-- BAD: Too many indexes on rarely-queried columns
CREATE INDEX idx_users_created_at ON users(created_at);
CREATE INDEX idx_users_updated_at ON users(updated_at);
CREATE INDEX idx_users_last_login ON users(last_login);
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_username ON users(username);

-- GOOD: Only index frequently-queried columns
CREATE INDEX idx_users_email ON users(email);  -- Used for login
CREATE INDEX idx_users_username ON users(username);  -- Used for lookup
-- Skip indexes on created_at, updated_at, last_login if rarely queried

7. Not Using Covering Indexes

Problem: Index scan followed by table lookup (heap fetch).

-- BAD: Index on id only, must fetch name from table
CREATE INDEX idx_users_email ON users(email);
SELECT id, name FROM users WHERE email = 'user@example.com';

-- GOOD: Include name in index (covering index)
CREATE INDEX idx_users_email_covering ON users(email) INCLUDE (name);
SELECT id, name FROM users WHERE email = 'user@example.com';
-- "Index Only Scan" - no heap fetch needed

8. String Pattern Matching Without Index

Problem: LIKE with leading wildcard can't use B-tree index.

-- BAD: Can't use standard index
CREATE INDEX idx_users_email ON users(email);
SELECT * FROM users WHERE email LIKE '%@example.com';  -- Seq Scan

-- GOOD: Use trigram index for pattern matching
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX idx_users_email_trgm ON users USING gin(email gin_trgm_ops);
SELECT * FROM users WHERE email LIKE '%@example.com';  -- Uses GIN index

Review Questions

1. Do all WHERE and JOIN columns have indexes? 2. Are composite index column orders optimized for queries? 3. Would partial indexes reduce index size and improve performance? 4. Are foreign keys indexed? 5. Has EXPLAIN ANALYZE been used to verify query plans? 6. Are there redundant or unused indexes? 7. Would covering indexes eliminate heap fetches? 8. Are pattern matching queries using appropriate index types (GIN, trigram)?

JSONB

Critical Anti-Patterns

1. Using JSON Instead of JSONB

Problem: JSON is stored as text, slower to query, no indexing.

-- BAD: Using JSON type
CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    metadata JSON
);

-- GOOD: Use JSONB for querying and indexing
CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    metadata JSONB
);

Rule: Always use JSONB unless you need to preserve exact formatting/whitespace.

2. Wrong JSONB Operator

Problem: -> returns JSONB, ->> returns text. Using wrong one breaks queries.

-- BAD: Comparing JSONB to text
SELECT * FROM users WHERE metadata->'age' = '25';  -- Won't work

-- GOOD: Use ->> for text comparison
SELECT * FROM users WHERE metadata->>'age' = '25';

-- GOOD: Use -> for JSONB comparison
SELECT * FROM users WHERE metadata->'age' = '25'::jsonb;

-- GOOD: Cast to integer for numeric comparison
SELECT * FROM users WHERE (metadata->>'age')::int = 25;

Operators:

• -> extracts as JSONB: metadata->'address' → {"city": "NYC"}
• ->> extracts as text: metadata->>'name' → "Alice"
• @> contains: metadata @> '{"role": "admin"}'
• ? key exists: metadata ? 'email'

3. Missing GIN Index on JSONB

Problem: JSONB queries without indexes perform sequential scans.

-- BAD: Querying JSONB without index
SELECT * FROM users WHERE metadata @> '{"role": "admin"}';  -- Seq Scan

-- GOOD: Create GIN index
CREATE INDEX idx_users_metadata ON users USING gin(metadata);
SELECT * FROM users WHERE metadata @> '{"role": "admin"}';  -- Uses index

-- GOOD: GIN index on specific path
CREATE INDEX idx_users_metadata_role ON users USING gin((metadata->'role'));

4. Not Using Containment Operator

Problem: Extracting and comparing is slower than using @>.

-- BAD: Extracting then comparing
SELECT * FROM events
WHERE metadata->>'type' = 'click' AND metadata->>'source' = 'mobile';

-- GOOD: Use containment operator
SELECT * FROM events
WHERE metadata @> '{"type": "click", "source": "mobile"}';
-- Much faster with GIN index

5. Storing Arrays as JSON Strings

Problem: Can't use array operators, must parse JSON every time.

# BAD: Storing array as JSON string
cursor.execute("""
    INSERT INTO users (tags) VALUES (%s)
""", (json.dumps(['python', 'postgres']),))

cursor.execute("""
    SELECT * FROM users WHERE tags::jsonb @> '"python"'
""")

# GOOD: Use PostgreSQL array type for simple arrays
cursor.execute("""
    INSERT INTO users (tags) VALUES (%s)
""", (['python', 'postgres'],))

cursor.execute("""
    SELECT * FROM users WHERE 'python' = ANY(tags)
""")

# Use JSONB only for complex nested structures

6. Deep Nesting Without Indexes

Problem: Querying deep paths is slow without expression indexes.

-- BAD: Querying deep path without index
SELECT * FROM events
WHERE metadata->'user'->'profile'->>'country' = 'US';

-- GOOD: Create expression index
CREATE INDEX idx_events_country ON events(
    (metadata->'user'->'profile'->>'country')
);
SELECT * FROM events
WHERE metadata->'user'->'profile'->>'country' = 'US';

7. Not Validating JSONB Structure

Problem: No schema validation leads to inconsistent data.

# BAD: No validation
cursor.execute("""
    INSERT INTO users (metadata) VALUES (%s)
""", (json.dumps({'age': 'twenty-five'}),))  # Should be integer!

# GOOD: Validate before insert
def validate_user_metadata(metadata: dict) -> dict:
    assert isinstance(metadata.get('age'), int), "age must be integer"
    assert isinstance(metadata.get('email'), str), "email must be string"
    return metadata

metadata = validate_user_metadata({'age': 25, 'email': 'user@example.com'})
cursor.execute("""
    INSERT INTO users (metadata) VALUES (%s)
""", (json.dumps(metadata),))

# BETTER: Use CHECK constraint
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    metadata JSONB,
    CHECK (jsonb_typeof(metadata->'age') = 'number'),
    CHECK (jsonb_typeof(metadata->'email') = 'string')
);

8. JSONB for Relational Data

Problem: Using JSONB when proper columns/foreign keys are better.

-- BAD: Storing relational data in JSONB
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    data JSONB  -- Contains user_id, product_id, quantity
);

-- GOOD: Use proper columns and foreign keys
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    product_id INTEGER REFERENCES products(id),
    quantity INTEGER,
    metadata JSONB  -- Only for truly unstructured data
);

Rule: Use JSONB for truly dynamic/unstructured data, not for avoiding schema design.

9. Inefficient JSONB Aggregation

Problem: Not using jsonb_agg or jsonb_object_agg.

# BAD: Fetching and building JSON in application code
cursor.execute("SELECT id, name FROM products WHERE category_id = %s", (cat_id,))
products = [{'id': row[0], 'name': row[1]} for row in cursor.fetchall()]
result = {'products': products}

# GOOD: Build JSON in database
cursor.execute("""
    SELECT jsonb_build_object(
        'products', jsonb_agg(jsonb_build_object('id', id, 'name', name))
    )
    FROM products
    WHERE category_id = %s
""", (cat_id,))
result = cursor.fetchone()[0]

Review Questions

1. Is JSONB used instead of JSON? 2. Are the correct operators used (-> vs ->>, @> for containment)? 3. Do JSONB columns have GIN indexes? 4. Are containment operators (@>) used instead of extracting and comparing? 5. Is JSONB used appropriately (not for relational data)? 6. Are deep paths indexed with expression indexes? 7. Is JSONB structure validated? 8. Are JSONB aggregation functions used instead of application-side building?

Transactions

Critical Anti-Patterns

1. Wrong Isolation Level

Problem: Using default isolation level when stronger guarantees needed.

# BAD: Default READ COMMITTED allows non-repeatable reads
async def transfer_money(from_id: int, to_id: int, amount: int):
    async with pool.acquire() as conn:
        async with conn.transaction():
            balance = await conn.fetchval(
                "SELECT balance FROM accounts WHERE id = $1", from_id
            )
            if balance < amount:
                raise ValueError("Insufficient funds")

            # Another transaction could modify balance here!
            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
            )

# GOOD: Use SERIALIZABLE for critical operations
async def transfer_money(from_id: int, to_id: int, amount: int):
    async with pool.acquire() as conn:
        async with conn.transaction(isolation='serializable'):
            balance = await conn.fetchval(
                "SELECT balance FROM accounts WHERE id = $1", from_id
            )
            if balance < amount:
                raise ValueError("Insufficient funds")

            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
            )

# BETTER: Use SELECT FOR UPDATE to lock row
async def transfer_money(from_id: int, to_id: int, amount: int):
    async with pool.acquire() as conn:
        async with conn.transaction():
            balance = await conn.fetchval(
                "SELECT balance FROM accounts WHERE id = $1 FOR UPDATE",
                from_id
            )
            if balance < amount:
                raise ValueError("Insufficient funds")

            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
            )

Isolation Levels:

• READ COMMITTED (default): Prevents dirty reads, allows non-repeatable reads
• REPEATABLE READ: Prevents dirty and non-repeatable reads
• SERIALIZABLE: Full isolation, prevents all anomalies

2. Long-Running Transactions

Problem: Holds locks, blocks other queries, bloats WAL.

# BAD: Long transaction holding locks
async def process_orders():
    async with pool.acquire() as conn:
        async with conn.transaction():
            orders = await conn.fetch("SELECT * FROM orders WHERE status = 'pending'")

            for order in orders:
                # External API call inside transaction!
                result = await external_api.process(order)

                await conn.execute(
                    "UPDATE orders SET status = $1, result = $2 WHERE id = $3",
                    'processed', result, order['id']
                )

# GOOD: Keep transactions short
async def process_orders():
    async with pool.acquire() as conn:
        orders = await conn.fetch("SELECT * FROM orders WHERE status = 'pending'")

    # Process outside transaction
    for order in orders:
        result = await external_api.process(order)

        async with pool.acquire() as conn:
            async with conn.transaction():
                await conn.execute(
                    "UPDATE orders SET status = $1, result = $2 WHERE id = $3",
                    'processed', result, order['id']
                )

3. Deadlocks from Lock Order

Problem: Different transactions acquire locks in different orders.

# BAD: Different lock order causes deadlocks
# Transaction 1
async with conn.transaction():
    await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")

# Transaction 2 (at same time)
async with conn.transaction():
    await conn.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 2")
    await conn.execute("UPDATE accounts SET balance = balance + 50 WHERE id = 1")
# DEADLOCK: T1 locks account 1, T2 locks account 2, both wait for each other

# GOOD: Always acquire locks in same order
async def transfer(from_id: int, to_id: int, amount: int):
    # Always lock lower ID first
    first_id, second_id = sorted([from_id, to_id])

    async with conn.transaction():
        # Lock in consistent order
        await conn.execute(
            "SELECT id FROM accounts WHERE id IN ($1, $2) ORDER BY id FOR UPDATE",
            first_id, second_id
        )

        if from_id < to_id:
            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
            )
        else:
            await conn.execute(
                "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
                amount, to_id
            )
            await conn.execute(
                "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                amount, from_id
            )

4. Not Using Advisory Locks

Problem: Application-level coordination requires database support.

# BAD: Race condition on external resource
async def process_unique_job(job_id: int):
    async with pool.acquire() as conn:
        job = await conn.fetchrow("SELECT * FROM jobs WHERE id = $1", job_id)

        if job['status'] == 'pending':
            # Another process could process same job!
            result = await expensive_operation(job)

            await conn.execute(
                "UPDATE jobs SET status = 'complete', result = $1 WHERE id = $2",
                result, job_id
            )

# GOOD: Use advisory lock
async def process_unique_job(job_id: int):
    async with pool.acquire() as conn:
        # Try to acquire advisory lock (non-blocking)
        locked = await conn.fetchval(
            "SELECT pg_try_advisory_lock($1)",
            job_id
        )

        if not locked:
            return  # Another process is handling this job

        try:
            job = await conn.fetchrow("SELECT * FROM jobs WHERE id = $1", job_id)

            if job['status'] == 'pending':
                result = await expensive_operation(job)

                await conn.execute(
                    "UPDATE jobs SET status = 'complete', result = $1 WHERE id = $2",
                    result, job_id
                )
        finally:
            # Release advisory lock
            await conn.execute("SELECT pg_advisory_unlock($1)", job_id)

Advisory Lock Functions:

• pg_advisory_lock(key): Blocking lock
• pg_try_advisory_lock(key): Non-blocking, returns true/false
• pg_advisory_unlock(key): Release lock
• pg_advisory_xact_lock(key): Auto-released at transaction end

5. Not Handling Serialization Failures

Problem: SERIALIZABLE transactions can fail and need retry.

# BAD: No retry on serialization failure
async def increment_counter(counter_id: int):
    async with pool.acquire() as conn:
        async with conn.transaction(isolation='serializable'):
            count = await conn.fetchval(
                "SELECT count FROM counters WHERE id = $1", counter_id
            )
            await conn.execute(
                "UPDATE counters SET count = $1 WHERE id = $2",
                count + 1, counter_id
            )
    # Raises SerializationError under contention

# GOOD: Retry on serialization failure
import asyncpg

async def increment_counter(counter_id: int, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            async with pool.acquire() as conn:
                async with conn.transaction(isolation='serializable'):
                    count = await conn.fetchval(
                        "SELECT count FROM counters WHERE id = $1", counter_id
                    )
                    await conn.execute(
                        "UPDATE counters SET count = $1 WHERE id = $2",
                        count + 1, counter_id
                    )
            return  # Success
        except asyncpg.SerializationError:
            if attempt == max_retries - 1:
                raise
            # Retry with exponential backoff
            await asyncio.sleep(0.1 * (2 ** attempt))

6. Missing ROLLBACK on Error

Problem: Transaction left open on error, holds locks.

# BAD: Transaction not rolled back on error
conn = pool.getconn()
cursor = conn.cursor()
cursor.execute("BEGIN")
try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    # Error here leaves transaction open!
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
finally:
    pool.putconn(conn)

# GOOD: Use context manager (auto rollback)
async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
        await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    # Automatically rolls back on exception

# GOOD: Explicit rollback
conn = pool.getconn()
try:
    cursor = conn.cursor()
    cursor.execute("BEGIN")
    try:
        cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
        cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
        conn.commit()
    except Exception:
        conn.rollback()
        raise
finally:
    pool.putconn(conn)

7. Nested Transactions Without Savepoints

Problem: Inner "transaction" doesn't actually create nested transaction.

# BAD: Nested transaction blocks don't work as expected
async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("INSERT INTO logs (message) VALUES ('start')")

        try:
            async with conn.transaction():  # This doesn't create nested transaction!
                await conn.execute("INSERT INTO data (value) VALUES (123)")
                raise ValueError("Error")
        except ValueError:
            pass  # Expect outer transaction to continue

        await conn.execute("INSERT INTO logs (message) VALUES ('end')")
# Entire transaction is rolled back, including 'start' log

# GOOD: Use savepoints for nested transactions
async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("INSERT INTO logs (message) VALUES ('start')")

        try:
            # Create savepoint
            await conn.execute("SAVEPOINT inner")
            await conn.execute("INSERT INTO data (value) VALUES (123)")
            raise ValueError("Error")
        except ValueError:
            # Rollback to savepoint
            await conn.execute("ROLLBACK TO SAVEPOINT inner")

        await conn.execute("INSERT INTO logs (message) VALUES ('end')")
# Both logs are committed, data insert is rolled back

8. Not Using FOR UPDATE SKIP LOCKED

Problem: Queue processing blocked by locked rows.

# BAD: Workers block on locked rows
async def process_next_job():
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Blocks if another worker locked this row
            job = await conn.fetchrow("""
                SELECT * FROM jobs
                WHERE status = 'pending'
                ORDER BY created_at
                LIMIT 1
                FOR UPDATE
            """)

            if job:
                await process_job(job)
                await conn.execute(
                    "UPDATE jobs SET status = 'complete' WHERE id = $1",
                    job['id']
                )

# GOOD: Skip locked rows
async def process_next_job():
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Skip rows locked by other workers
            job = await conn.fetchrow("""
                SELECT * FROM jobs
                WHERE status = 'pending'
                ORDER BY created_at
                LIMIT 1
                FOR UPDATE SKIP LOCKED
            """)

            if job:
                await process_job(job)
                await conn.execute(
                    "UPDATE jobs SET status = 'complete' WHERE id = $1",
                    job['id']
                )

Review Questions

1. Is the isolation level appropriate for the operation? 2. Are transactions kept short (no I/O inside)? 3. Are locks always acquired in consistent order to prevent deadlocks? 4. Would advisory locks help with application-level coordination? 5. Are serialization failures caught and retried? 6. Are transactions properly rolled back on error (context managers)? 7. Are savepoints used for nested transaction semantics? 8. Is FOR UPDATE SKIP LOCKED used for queue processing?