PostgreSQL in Production

Postgres is the default relational database for production Python applications in 2026, and the reasons compound. Full SQL with a sane standards-compliance record. JSONB columns that have absorbed most of what document stores used to be for. Full-text search built in. Geospatial via PostGIS. Real MVCC so readers don't block writers. A driver ecosystem — psycopg, asyncpg, SQLAlchemy on top of either — that is mature, fast, and well-maintained.

This lesson is the production-shaped tour: drivers, pooling, transactions, the Postgres-specific data types worth knowing, async, migrations, and the performance levers you reach for before any of it matters at scale.

Examples assume a running local Postgres — Docker is the easiest way; see devops-docker. Pip-install the drivers as shown. Expected output is in comments.

1. The Driver Landscape

Four names you will hear; pick deliberately.

Driver	Sync	Async	Notes
`psycopg` (v3)	yes	yes	Modern, recommended default. Same package handles both.
`psycopg2`	yes	no	Legacy, still widespread; fine for existing code.
`asyncpg`	no	yes	Pure async, fastest, no built-in SQLAlchemy bridge.
SQLAlchemy 2.0	yes	yes	Sits on top of any of the above. Recommended for most apps.

If you're starting fresh and want sync — psycopg 3. If you're writing a FastAPI service that needs raw speed — asyncpg, or psycopg 3 in async mode. If you want an ORM — SQLAlchemy 2.0 (see orm-patterns) with psycopg 3 as the DBAPI.

psycopg2 is not deprecated, but new code should target psycopg 3 — better connection pooling, async support, and a cleaner API.

2. Connecting with `psycopg` 3

bash

pip install "psycopg[binary]"

The [binary] extra ships a pre-built wheel — no system Postgres client needed. For production Linux servers, prefer psycopg[c] and the system libpq.

python

import psycopg

with psycopg.connect("postgresql://app:secret@localhost:5432/myapp") as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT id, email FROM users WHERE active = %s", (True,))
        for row in cur:
            print(row)
# Expected: tuples for each active user, e.g. (1, 'alice@example.com')

Five things to internalise:

The connection is a context manager — with conn: commits on success, rolls back on exception.
The cursor is also a context manager — closes when you leave the block.
%s is the parameter placeholder, not a Python format specifier. psycopg parses it. There is no %d, no %f. Everything is %s.
The cursor is iterable — for row in cur: streams rows on demand. No fetchall() into a list unless you need all of them in memory.
The connection string honours PG* environment variables (PGHOST, PGUSER, PGPASSWORD, PGDATABASE) — read these via envconfig and never hardcode credentials.

3. Parameterised Queries — The One Non-Negotiable

This is the same warning as every database lesson, and it is the same warning because the bug never stops happening. Never build SQL by f-string or + concatenation.

python

# CATASTROPHIC
email = request.form["email"]
cur.execute(f"SELECT id FROM users WHERE email = '{email}'")

+ setup added so this can run · defines request, cur

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

request = _AutoMock('request')
cur = _AutoMock('cur')

A user typing ' OR 1=1; -- dumps your entire users table. The fix is one character:

python

# CORRECT
cur.execute("SELECT id FROM users WHERE email = %s", (email,))

+ setup added so this can run · defines cur, email

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')
email = _AutoMock('email')

For IN (...) lists, use ANY:

python

cur.execute("SELECT id FROM users WHERE id = ANY(%s)", ([1, 2, 3],))

+ setup added so this can run · defines cur

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')

For dynamic identifiers (table names, column names — which cannot be parameterised), use psycopg.sql.Identifier:

python

from psycopg import sql
cur.execute(
    sql.SQL("SELECT * FROM {tbl} WHERE id = %s").format(tbl=sql.Identifier(table_name)),
    (user_id,),
)

+ setup added so this can run · defines cur, user_id, table_name

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')
user_id = _AutoMock('user_id')
table_name = _AutoMock('table_name')

See security-checklist for the wider injection picture.

4. Connection Pooling — Non-Negotiable in Production

A fresh TCP + TLS + auth handshake to Postgres takes 20-50 ms. A web request that opens and closes a connection per request gives up most of its latency budget to setup. Use a pool.

python

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo="postgresql://app:secret@localhost/myapp",
    min_size=4,
    max_size=20,
    timeout=10,           # seconds to wait for a free conn before raising
    max_lifetime=3600,    # recycle connections after an hour
    open=True,
)

def get_user(user_id):
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, email FROM users WHERE id = %s", (user_id,))
            return cur.fetchone()

pool.connection() borrows a connection, runs the block, and returns it. Sizing rule of thumb: max_size per process ≈ (target concurrent requests) / (process count), capped at what Postgres can handle (max_connections, usually 100 by default — raise it or run PgBouncer in front).

For multi-process deployments (Gunicorn workers, Celery), each worker gets its own pool. Don't try to share a pool across processes.

5. Transactions

psycopg defaults to a transaction-per-block model. The connection is a context manager: exit cleanly → commit; exit on exception → rollback.

python

with pool.connection() as conn:
    with conn.transaction():
        conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
        conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))
        # commits here; rolls back if anything raises

+ setup added so this can run · defines pool

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

pool = _AutoMock('pool')

conn.transaction() is an explicit savepoint-aware transaction block. Nesting it gives you savepoints, useful when one optional step is allowed to fail without aborting the whole transaction:

python

with conn.transaction():
    conn.execute("INSERT INTO orders ...")
    try:
        with conn.transaction():           # savepoint
            conn.execute("INSERT INTO loyalty_points ...")
    except Exception:
        pass                               # outer transaction still commits

+ setup added so this can run · defines conn

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

conn = _AutoMock('conn')

Keep transactions short. A transaction holds row-level locks; long-running ones block other writers and bloat MVCC version chains.

6. Postgres-Specific Types Worth Knowing

The reason Postgres beats SQLite for serious work: a real type system.

JSONB — binary JSON, indexable and queryable:

python

cur.execute("INSERT INTO events (payload) VALUES (%s)", ({"type": "click", "x": 42},))

cur.execute("SELECT id FROM events WHERE payload->>'type' = %s", ("click",))
cur.execute("SELECT id FROM events WHERE payload @> %s", ('{"type": "click"}',))

+ setup added so this can run · defines cur

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')

->> extracts a field as text. @> is "JSON contains". A GIN index on payload makes containment queries fast:

sql

CREATE INDEX idx_events_payload ON events USING GIN (payload);

ARRAY — first-class arrays of any type:

python

cur.execute("INSERT INTO posts (tags) VALUES (%s)", (["python", "postgres"],))
cur.execute("SELECT id FROM posts WHERE %s = ANY(tags)", ("python",))

+ setup added so this can run · defines cur

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

cur = _AutoMock('cur')

UUID — uuid type for primary keys; pair with gen_random_uuid() server-side.

TIMESTAMPTZ — always use timestamps with time zone. Storing TIMESTAMP WITHOUT TIME ZONE is the source of half the daylight-savings bugs in this industry. psycopg returns aware datetime objects automatically.

ENUM — typed enums at the column level; combine with Python enum.Enum for safety.

7. Server-Side Cursors for Huge Result Sets

A normal cur.execute("SELECT * FROM big_table") pulls every row into client memory. For a billion-row table that's a heap exhaustion waiting to happen. Named cursors stream rows in batches:

python

with conn.cursor(name="exporter") as cur:
    cur.itersize = 1000                     # fetch 1000 at a time
    cur.execute("SELECT id, payload FROM big_table")
    for row in cur:
        process(row)
# Memory stays flat regardless of table size.

+ setup added so this can run · defines conn, process

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

conn = _AutoMock('conn')
def process(*_a, **_kw):
    print('-> process() called')
    return _AutoMock('process()')

Naming the cursor flips it to a server-side cursor — Postgres holds the result set and ships rows on demand. Critical for ETL, exports, and any reporting query that doesn't fit in RAM.

8. Async with `asyncpg`

When you're inside an asyncio application — FastAPI, an async worker, an async scraper (see async) — block the loop and you destroy your throughput. Use an async driver.

bash

pip install asyncpg

python

import asyncpg
import asyncio

async def main():
    conn = await asyncpg.connect("postgresql://app:secret@localhost/myapp")
    try:
        rows = await conn.fetch("SELECT id, email FROM users WHERE active = $1", True)
        for r in rows:
            print(r["id"], r["email"])
    finally:
        await conn.close()

asyncio.run(main())

Notice the placeholder: $1, $2 — not %s. asyncpg uses Postgres-native positional placeholders. Rows are asyncpg.Record objects — dict-like and tuple-like.

Pooling is mandatory the moment you have concurrent requests:

python

pool = await asyncpg.create_pool(
    dsn="postgresql://app:secret@localhost/myapp",
    min_size=4,
    max_size=20,
)

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

asyncpg is roughly 2-3× faster than psycopg 3's async mode for raw throughput; psycopg 3 wins on ergonomics and SQLAlchemy compatibility. Pick once per service.

9. Migrations

The schema evolves. Migrations make that evolution reproducible.

Alembic — standalone, works with SQLAlchemy. The default for non-Django Python apps. See orm-patterns.
Django migrations — built into Django. See django-models.
Flask-Migrate — Alembic wrapper for Flask. See flask-database.
Raw SQL files + a runner (yoyo-migrations, dbmate) — when you want SQL, not Python.

The one rule that matters across all of them: migrations are append-only. Once a migration is applied to production, you don't edit it. You write a new migration that undoes or amends.

Backup before destructive migrations: pg_dump -Fc myapp > myapp.dump; restore with pg_restore -d myapp myapp.dump.

10. Performance Basics

Two tools cover 90% of slow-query investigations.

EXPLAIN ANALYZE — runs the query and shows the actual plan:

sql

EXPLAIN ANALYZE SELECT id FROM events WHERE payload->>'type' = 'click';

Look for Seq Scan on big tables — that's a full table scan, almost always fixable with an index. Look for huge rows × loops products — that's nested-loop trouble.

Indexes — the right index turns a 5-second query into a 5-millisecond query.

Index	Use case
B-tree (default)	Equality and range on scalars
GIN	JSONB containment, array membership, full-text search
GiST	Geospatial, range types
BRIN	Huge tables ordered by insertion time (logs, events)
Hash	Equality only; rarely worth it over B-tree

Index every foreign key. Index every column you frequently WHERE on. Don't index every column — writes pay the cost of every index.

VACUUM runs automatically (autovacuum). If it falls behind on a write-heavy table, tune autovacuum_vacuum_scale_factor lower. Symptoms of autovacuum trouble: bloat, slow sequential scans, transaction-ID wraparound warnings.

PgBouncer sits between your app and Postgres, multiplexing many client connections onto few server connections. Once you have more than ~50 app processes, you need it.

11. When NOT to Use Postgres

Postgres is the default, not the answer to everything.

Caching — Redis is 10-100× faster for read-heavy hot data. See db-redis.
Schemaless documents — JSONB covers most cases; reach for db-mongo only when the document model truly fits.
Time-series at huge scale — TimescaleDB (Postgres extension) is great; ClickHouse / InfluxDB win at petabyte scale.
Full-text search at scale — Postgres FTS is fine for millions of rows; Elasticsearch wins beyond that.
Analytics over hundreds of GB — column stores (ClickHouse, DuckDB, BigQuery) crush Postgres for OLAP workloads.

Common Mistakes

1. Building SQL with f-strings. Same warning, every lesson. Use %s placeholders (psycopg) or $1 (asyncpg). No exceptions.

2. No connection pool. Every web request opens a fresh connection → 30 ms of handshake per request → throughput floor of ~30 RPS per process. Always pool.

3. Long-running transactions. A transaction that stays open for minutes holds locks and bloats MVCC. Open late, commit early. Never BEGIN and then go do a slow external API call.

4. SELECT * then filter in Python. Pull the rows you need, with the columns you need, ordered and limited at the database. SELECT * plus a Python loop on a million-row table is wasted network and memory.

5. Ignoring EXPLAIN ANALYZE. If a query is slow, look at the plan before guessing. The fix is almost always an index, a query rewrite, or denormalisation — and the plan tells you which.

6. Running migrations from one server while another serves writes. A schema-changing migration takes locks. Two servers racing to apply migrations deadlock or apply twice. One deploy process applies migrations; the rest start once migrations finish.

7. Storing timestamps without a time zone. Use TIMESTAMPTZ. Always. The first DST event will teach you why.

🎯 Your Turn — A Pooled `UserRepo`

Build a UserRepo class that wraps a psycopg_pool.ConnectionPool and exposes two operations: get_by_email and create. Use parameterised queries throughout, return None for the missing case, and return the new row's id from create.

Assume a table:

sql

CREATE TABLE users (
    id       BIGSERIAL PRIMARY KEY,
    email    TEXT NOT NULL UNIQUE,
    name     TEXT NOT NULL,
    created  TIMESTAMPTZ NOT NULL DEFAULT now()
);

Skeleton:

python

from psycopg_pool import ConnectionPool

class UserRepo:
    def __init__(self, dsn: str, *, min_size: int = 2, max_size: int = 10):
        # TODO 1: create the ConnectionPool and store it on self
        ...

    def get_by_email(self, email: str) -> dict | None:
        # TODO 2: parameterised SELECT; return dict or None
        ...

    def create(self, email: str, name: str) -> int:
        # TODO 3: INSERT ... RETURNING id; return the new id
        ...

    def close(self):
        # TODO 4: close the pool (for clean shutdown)
        ...

Hint 1 — Dict rows

Pass row_factory=psycopg.rows.dict_row to the cursor (or set it on the connection) so fetchone() returns a dict instead of a tuple. Import from psycopg.rows.

Hint 2 — RETURNING

Postgres lets you append RETURNING id to an INSERT and read back the generated key in the same round trip — no separate SELECT needed. cur.fetchone()[0] gives you the int.

Show full solution

python

import psycopg
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool


class UserRepo:
    def __init__(self, dsn: str, *, min_size: int = 2, max_size: int = 10):
        self.pool = ConnectionPool(
            conninfo=dsn,
            min_size=min_size,
            max_size=max_size,
            timeout=10,
            open=True,
        )

    def get_by_email(self, email: str) -> dict | None:
        with self.pool.connection() as conn:
            with conn.cursor(row_factory=dict_row) as cur:
                cur.execute(
                    "SELECT id, email, name, created FROM users WHERE email = %s",
                    (email,),
                )
                return cur.fetchone()

    def create(self, email: str, name: str) -> int:
        with self.pool.connection() as conn:
            with conn.transaction():
                with conn.cursor() as cur:
                    cur.execute(
                        "INSERT INTO users (email, name) VALUES (%s, %s) RETURNING id",
                        (email, name),
                    )
                    (new_id,) = cur.fetchone()
                    return new_id

    def close(self):
        self.pool.close()


# Demo
repo = UserRepo("postgresql://app:secret@localhost/myapp")
try:
    new_id = repo.create("alice@example.com", "Alice")
    print("created:", new_id)                   # created: 1
    print("lookup :", repo.get_by_email("alice@example.com"))
    # lookup : {'id': 1, 'email': 'alice@example.com', 'name': 'Alice', 'created': datetime(...)}
finally:
    repo.close()

What the solution gets right:

Pooling — one pool per repo, reused across calls. No reconnection per query.
Parameterised queries everywhere — no f-strings, no string concatenation.
RETURNING id — one round trip, atomic with the insert.
dict_row factory — callers get dicts, not tuples; rename a column and they still work.
Explicit transaction on writes — with conn.transaction(): makes the commit boundary obvious.
close() — drains the pool cleanly on shutdown; pair with a FastAPI/Flask lifespan hook.

For an async version, swap psycopg_pool.ConnectionPool for psycopg_pool.AsyncConnectionPool (or asyncpg.create_pool), make every method async, and await the pool/cursor calls. The shape is identical.

What You Learned

psycopg 3 is the default modern Postgres driver — sync and async in one package.
Parameterised queries with %s (psycopg) or $1 (asyncpg). Never f-strings.
Connection pooling via psycopg_pool.ConnectionPool or asyncpg.create_pool is non-negotiable in production.
Transactions with conn.transaction() — short, scoped, no slow work inside.
Postgres-native types — JSONB with GIN indexes, ARRAY, UUID, TIMESTAMPTZ, ENUM.
Server-side cursors for huge result sets — cursor(name=...) streams in batches.
asyncpg for raw async throughput; psycopg 3 async for ergonomics + SQLAlchemy.
Migrations with Alembic / Django / Flask-Migrate — append-only, one applier per deploy.
EXPLAIN ANALYZE and indexes (B-tree, GIN, BRIN) cover most performance work.
PgBouncer when you outgrow Postgres's connection limit.
Not for everything — Redis for cache, Mongo for document-shaped data, OLAP stores for analytics.

Next: Redis Patterns — cache, sessions, queues, rate limits, distributed locks.

1. The Driver Landscape

2. Connecting with psycopg 3

3. Parameterised Queries — The One Non-Negotiable

4. Connection Pooling — Non-Negotiable in Production

5. Transactions

6. Postgres-Specific Types Worth Knowing

7. Server-Side Cursors for Huge Result Sets

8. Async with asyncpg

9. Migrations

10. Performance Basics

11. When NOT to Use Postgres

Common Mistakes

🎯 Your Turn — A Pooled UserRepo

What You Learned

2. Connecting with `psycopg` 3

8. Async with `asyncpg`

🎯 Your Turn — A Pooled `UserRepo`