PostgreSQL in Production

Postgres is the default relational database for production Python applications in 2026, and there are several reasons. It has full SQL and follows the SQL standard well. Its JSONB columns now handle most of the jobs that people used document stores for. It has full-text search built in, and geospatial support via PostGIS. It has real MVCC, so readers don't block writers. And its Python drivers (psycopg, asyncpg, and SQLAlchemy on top of either) are mature, fast and well maintained.

This lesson covers what you need in production: drivers, pooling, transactions, the Postgres-specific data types worth knowing, async, migrations, and the first performance tools to use as your app grows.

The examples assume a running local Postgres. Docker is the easiest way to get one. See devops-docker. Pip-install the drivers as shown. Expected output is in comments.

1. The Drivers

You will hear four names. Choose carefully.

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 code, use psycopg 3. If you're writing a FastAPI service that needs raw speed, use asyncpg, or psycopg 3 in async mode. If you want an ORM, use SQLAlchemy 2.0 (see orm-patterns) with psycopg 3 as the DBAPI.

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

2. Connecting with `psycopg` 3

bash

pip install "psycopg[binary]"

The [binary] extra installs a pre-built wheel, so you don't need a system Postgres client. 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: one tuple per active user, for example (1, 'alice@example.com')

Five things to remember:

The connection is a context manager: with conn: commits on success and rolls back on exception.
The cursor is also a context manager: it 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: gives you one row at a time. Don't fetchall() into a list unless you need all the rows 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 Rule You Must Never Break

Every database lesson repeats this warning, because this bug keeps happening. Never build SQL with an 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 who types ' OR 1=1; -- gets your entire users table. The fix is a small change:

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 and 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: Required in Production

A fresh TCP + TLS + auth handshake to Postgres takes 20-50 ms. If your app opens and closes a new connection for every web request, most of the request's time goes into 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 connection before raising an error
    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), but no more than Postgres can handle (max_connections, usually 100 by default, so 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, and 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():           # a nested transaction (a 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 make MVCC keep many old row versions (bloat).

6. Postgres-Specific Types Worth Knowing

One reason Postgres is better than SQLite for serious work: a real type system.

JSONB: binary JSON that you can index and query:

python

from psycopg.types.json import Jsonb

cur.execute("INSERT INTO events (payload) VALUES (%s)", (Jsonb({"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: real 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 it with gen_random_uuid() server-side.

TIMESTAMPTZ: always use timestamps with time zone. Storing TIMESTAMP WITHOUT TIME ZONE is a very common cause of daylight-saving-time bugs. psycopg returns aware datetime objects automatically.

ENUM: typed enums at the column level. Combine them 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 will run out of memory. 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 use stays the same, however big the table is.

+ 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()')

Giving the cursor a name makes it a server-side cursor: Postgres keeps the result set and sends rows as you ask for them. This is essential for ETL, exports, and any reporting query whose result doesn't fit in RAM.

8. Async with `asyncpg`

Inside an asyncio application (FastAPI, an async worker, an async scraper), a blocking database call stops the event loop, and throughput drops badly (see async). 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, which work like both a dict and a tuple.

As soon as you have concurrent requests, you need a pool:

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, but psycopg 3 is easier to use and works better with SQLAlchemy. 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 for 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 most 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, and an index almost always fixes it. Look for huge rows × loops products: that means a slow nested loop.

Indexes: the right index can turn 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, because every index makes writes slower.

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, but it is not the best tool for everything.

Caching: Redis is much faster for frequently read data. See db-redis.
Schemaless documents: JSONB covers most cases. Use db-mongo only when the document model really fits.
Time-series at huge scale: TimescaleDB (Postgres extension) is great, but ClickHouse / InfluxDB win at petabyte scale.
Full-text search at scale: Postgres FTS is fine for millions of rows, but Elasticsearch is better beyond that.
Analytics over hundreds of GB: column stores (ClickHouse, DuckDB, BigQuery) are far faster than Postgres for OLAP workloads.

Common Mistakes

1. Building SQL with f-strings. The same warning as in 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 → at most ~30 RPS per process. Always use a pool.

3. Long-running transactions. A transaction that stays open for minutes holds locks and bloats MVCC. Open the transaction as late as possible and commit as early as possible. Never BEGIN and then make 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. If two servers try to apply migrations at the same time, they can deadlock or apply a migration twice. One deploy process applies migrations, and the rest start once migrations finish.

7. Storing timestamps without a time zone. Always use TIMESTAMPTZ. Otherwise you will get bugs the first time the clocks change for daylight saving time.

🎯 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: run a parameterised SELECT and return a dict or None
        ...

    def create(self, email: str, name: str) -> int:
        # TODO 3: run INSERT ... RETURNING id and 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, with no separate SELECT. 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, so their code still works if you reorder the columns.
Explicit transaction on writes: with conn.transaction(): makes the commit boundary obvious.
close(): closes the pool cleanly on shutdown. Pair it 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 structure is the same.

What You Learned

psycopg 3 is the default modern Postgres driver, with 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 required 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, and 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 Drivers

2. Connecting with psycopg 3

3. Parameterised Queries: The One Rule You Must Never Break

4. Connection Pooling: Required 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`