Cheatsheet for connecting SQLAlchemy to Postgres. Long-form: Postgres textbook Ch 1 .

Driver matrix

DriverSync/AsyncUse
asyncpgasyncDefault for async; fastest
psycopg (v3)bothNative async; closer to libpq
psycopg2syncLegacy
pg8000syncPure Python; libpq-free

URL

"postgresql+asyncpg://user:pass@host:5432/db"
"postgresql+psycopg://user:pass@host:5432/db"          # v3
"postgresql+psycopg2://user:pass@host:5432/db"

connect_args (asyncpg)

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

connect_args (psycopg v3)

engine = create_async_engine(URL, connect_args={
    "options": "-c search_path=myapp,public -c application_name=myapp -c timezone=UTC",
})

PgBouncer (transaction-pool mode) — required tweaks

engine = create_async_engine(
    "postgresql+asyncpg://app@pgbouncer:6432/db",
    pool_size=10,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=300,
    connect_args={
        "statement_cache_size": 0,             # required
        "prepared_statement_cache_size": 0,
        "server_settings": {"jit": "off"},
    },
)

Without these: random “prepared statement does not exist” errors.

TLS

engine = create_async_engine(URL, connect_args={
    "ssl": "require",   # asyncpg: "disable" | "allow" | "prefer" | "require" | "verify-ca" | "verify-full"
})

For verify-full:

import ssl
ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
engine = create_async_engine(URL, connect_args={"ssl": ctx})

RDS IAM auth

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

Token refreshes per connection.

Cloud SQL Auth Proxy

Run proxy as a sidecar; connect to localhost.

application_name (debugging)

SELECT pid, application_name, state, query
FROM pg_stat_activity
WHERE application_name = 'myapp';

Each app’s connections visible. Set distinct names per service.

search_path per session

@event.listens_for(engine.sync_engine, "connect")
def on_connect(dbapi_conn, conn_record):
    cur = dbapi_conn.cursor()
    cur.execute("SET search_path TO myapp_schema, public")
    cur.close()

For schema-per-tenant: rotate per request via SET LOCAL.

statement_timeout

# Per-connection (durable)
@event.listens_for(engine.sync_engine, "connect")
def on_connect(dbapi_conn, conn_record):
    cur = dbapi_conn.cursor()
    cur.execute("SET statement_timeout = '30s'")
    cur.close()

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

idle_in_transaction_session_timeout

ALTER SYSTEM SET idle_in_transaction_session_timeout = '60s';

Kills connections holding open transactions while idle. Prevents lock pile-ups.

Read replicas

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

Multi-host failover

"postgresql+asyncpg://app@primary:5432,replica:5432/db?target_session_attrs=read-write"

asyncpg picks the writable host; handles failover.

Patroni / Stolon

App connects to a stable endpoint (HAProxy / pgcat). Patroni manages failover behind it.

Verify pool

print(engine.pool.size(), engine.pool.checked_out())

Expose as Prometheus gauge.

Common mistakes

  • Mixing psycopg2 with async engine — fails.
  • PgBouncer transaction-pool without disabling prepared-stmt cache.
  • TLS without server cert verification — MITM risk.
  • Generic application_name (or unset) — debugging nightmares.
  • Forgetting statement_timeout — runaway queries.

Read this next

If you want my Postgres engine + RDS IAM template, it’s at rajpoot.dev .

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 .