This is Chapter 1 of the Postgres-focused SQLAlchemy textbook. Pair with the DB-agnostic textbook for fundamentals; this volume goes deep on Postgres-specific dialect features.

Driver options

DriverAsyncNotes
asyncpgasyncFastest; pure-Python; widely used
psycopg (v3)bothNative async support; uses libpq
psycopg2syncMature; the legacy default
pg8000syncPure Python; used when libpq unavailable

For 2026 production async: asyncpg is the default. psycopg v3 catches up with broader feature support; closer to libpq.

Engine setup

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

# psycopg v3 (sync)
engine = create_engine("postgresql+psycopg://user:pass@host/db")

# psycopg v3 (async)
engine = create_async_engine("postgresql+psycopg://user:pass@host/db")

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

Connection args

engine = create_async_engine(
    URL,
    connect_args={
        "server_settings": {
            "application_name": "myapp",
            "search_path": "myapp_schema,public",
            "timezone": "UTC",
        },
        "command_timeout": 60,
        "statement_cache_size": 1000,
    },
)

asyncpg-specific. For psycopg v3:

connect_args={
    "options": "-c search_path=myapp_schema,public -c application_name=myapp",
}

Different syntax; same intent.

Why application_name matters

SELECT pid, application_name, state, query FROM pg_stat_activity;

Each app’s connections show up named. Vital for debugging “who’s holding this connection?”

Set per app — and per role if you have multiple roles.

search_path

Postgres looks up unqualified table names in search_path. Default: "$user", public.

connect_args={"server_settings": {"search_path": "myapp,public"}}

For schema-per-tenant or namespaced apps.

SSL / TLS

engine = create_async_engine(
    URL + "?ssl=require",
    connect_args={
        "ssl": "require",  # or "verify-ca", "verify-full"
    },
)

For production: at minimum require. Better: verify-full with CA certs.

PgBouncer mode

For PgBouncer in transaction-pooling:

connect_args={
    "statement_cache_size": 0,  # asyncpg
    "prepared_statement_cache_size": 0,
    "server_settings": {"jit": "off"},  # avoid prepared-stmt JIT issues
}

Without disabling stmt cache, you’ll see weird errors as PgBouncer hands different sessions different connections.

Connection pooling alongside PgBouncer

engine = create_async_engine(
    "postgresql+asyncpg://app@pgbouncer:6432/db",
    pool_size=10,
    max_overflow=10,
    pool_pre_ping=True,
)

App-level pool + PgBouncer = layered. PgBouncer absorbs the multiplexing.

Setting per-session config

@event.listens_for(engine.sync_engine, "connect")
def set_config(dbapi_conn, conn_record):
    cursor = dbapi_conn.cursor()
    cursor.execute("SET TIME ZONE 'UTC'")
    cursor.execute("SET statement_timeout = '30s'")
    cursor.close()

Runs once per physical connection. For per-session: per-request via Depends + SET LOCAL.

Statement timeout

cursor.execute("SET statement_timeout = '5s'")

Postgres kills queries running longer than this. Critical safety net.

For per-query:

async with session.begin():
    await session.execute(text("SET LOCAL statement_timeout = '60s'"))
    await session.execute(slow_stmt)

SET LOCAL resets at transaction end.

idle_in_transaction_session_timeout

ALTER SYSTEM SET idle_in_transaction_session_timeout = '60s';

Kills connections holding open transactions while idle. Catches forgotten transactions; saves locks.

RDS / Cloud SQL specifics

  • AWS RDS: IAM auth, parameter groups, IAM-DB-auth tokens.
  • GCP Cloud SQL: connection via Cloud SQL Auth Proxy.
  • Aurora: similar to RDS; multi-AZ failover automatic.

For IAM auth, generate token per connection:

@event.listens_for(engine.sync_engine, "do_connect")
def fetch_token(dialect, conn_rec, cargs, cparams):
    cparams["password"] = generate_iam_token()

Read replicas

write_engine = create_async_engine(WRITER_URL)
read_engine = create_async_engine(READER_URL, connect_args={"server_settings": {"default_transaction_read_only": "on"}})

Read engine forced read-only at session start. Writes fail loudly.

Connection per-process scaling

Workers × replicas × pool = total Postgres connections

Postgres max_connections default 100. PgBouncer multiplexes:

[App pods (50) × pool 5] = 250 logical
[PgBouncer pool to PG] = 50 actual

Async event loop and asyncpg

asyncpg ties connections to event loops. Don’t share across loops. In FastAPI: one event loop per worker; fine.

Pool-handled retries

engine = create_async_engine(URL, pool_pre_ping=True)

Pre-ping detects dead connections; replaces them. Without, requests fail after Postgres restarts / failovers.

What this textbook covers

ChTopic
1This chapter — driver setup
2Postgres types: arrays, JSONB, ranges, citext, intervals
3Indexes: B-tree, GIN, GiST, BRIN, HASH, partial, expression
4Full-text search
5Locking and advisory locks
6LISTEN / NOTIFY for pub/sub
7Partitioning
8RLS for multi-tenancy
9Postgres-specific upsert and bulk ops
10Replication, failover, pgvector

Read this next

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .