Configuring connect_args and server_settings for asyncpg

Put SQLAlchemy's own options in engine keyword arguments, asyncpg's in connect_args (ssl, timeout, command_timeout), and PostgreSQL run-time parameters in connect_args["server_settings"] (application_name, statement_timeout, timezone) — a setting in the wrong layer either raises TypeError or is silently ignored. This guide belongs to configuring async engines and connection pools.

Quick Answer

Three layers of configuration look interchangeable and are not. libpq-style names do not reach asyncpg, and PostgreSQL parameters do not belong at the top level of connect_args.

Before — settings in the wrong layers:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    connect_args={
        "sslmode": "require",                  # libpq name: asyncpg has no such argument
        "connect_timeout": 10,                 # libpq name
        "application_name": "orders-api",      # a PostgreSQL parameter, not an asyncpg one
        "options": "-c statement_timeout=5000",
    },
)
# TypeError: connect() got an unexpected keyword argument 'sslmode'

After — each setting in the layer that understands it:

import ssl

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

ssl_context = ssl.create_default_context()

engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    # SQLAlchemy's own options.
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
    pool_recycle=1800,
    # asyncpg's connect() options.
    connect_args={
        "ssl": ssl_context,
        "timeout": 10,             # seconds to establish the connection
        "command_timeout": 30,     # client-side ceiling per command
        # PostgreSQL run-time parameters, sent at start-up.
        "server_settings": {
            "application_name": "orders-api",
            "statement_timeout": "5s",
            "idle_in_transaction_session_timeout": "30s",
            "timezone": "UTC",
        },
    },
)
Session = async_sessionmaker(engine, expire_on_commit=False)

Every value in server_settings must be a string — asyncpg sends them verbatim as start-up parameters — and PostgreSQL accepts interval strings like "5s" as well as bare milliseconds for the timeout settings.

Execution Context & Async Workflow Integration

The three layers correspond to three different pieces of software.

Engine keyword arguments configure SQLAlchemy: how many connections to pool, when to recycle them, whether to ping before handing one out, whether to echo statements. They never reach the driver.

connect_args is passed through to asyncpg.connect(). Its names are asyncpg's, which is why libpq spellings fail: ssl rather than sslmode, timeout rather than connect_timeout, and no options parameter at all. The full set worth knowing is small — ssl, timeout, command_timeout, server_settings, statement_cache_size, prepared_statement_cache_size, prepared_statement_name_func — and migrating from psycopg2 to asyncpg lists the translations.

server_settings is a dictionary asyncpg sends as PostgreSQL start-up parameters. They take effect before the first statement, with no extra round trip, and apply for the life of the connection. Anything settable with SET is settable here, which makes it the right place for per-service defaults.

The two timeouts are worth separating clearly, because they fail differently. command_timeout is enforced in the client: asyncpg stops waiting, cancels the query and raises, which bounds how long your task is blocked. statement_timeout is enforced by PostgreSQL: the server stops executing and returns an error, which bounds how long the database works. A client-side timeout alone can leave the server still executing; a server-side timeout alone leaves your task waiting for a cancellation round trip. Setting both, with the client value slightly higher, covers each case.

idle_in_transaction_session_timeout is the one most often missing, and the most valuable. A session that opens a transaction and then stops working — a leaked session, a task cancelled at the wrong moment, a developer's psql window — holds its locks and prevents vacuum from cleaning up rows for as long as it lives. Thirty seconds turns that into a terminated backend rather than an incident, and it is exactly the failure described in detecting idle-in-transaction sessions from async code.

One caution about server_settings: search_path belongs there only for a single-tenant service. For schema-per-tenant routing, the connection is pooled and shared, so the schema has to change per request — which is what schema_translate_map is for, as described in switching schemas per request.

Resolving Warnings, Errors & Common Mistakes

Verifying that the settings actually arrived takes one query, and it is worth doing in a startup check rather than assuming — a typo in a server_settings key is accepted by PostgreSQL as a custom parameter and silently does nothing:

from sqlalchemy import text

EXPECTED = {
    "application_name": "orders-api",
    "statement_timeout": "5s",
    "idle_in_transaction_session_timeout": "30s",
    "TimeZone": "UTC",
}


async def verify_server_settings(engine) -> None:
    async with engine.connect() as conn:
        rows = await conn.execute(text(
            "SELECT name, setting FROM pg_settings WHERE name = ANY(:names)"
        ), {"names": list(EXPECTED)})
        actual = {name: setting for name, setting in rows}
    for name, expected in EXPECTED.items():
        if name not in actual:
            raise RuntimeError(f"{name} was not applied — check the server_settings key")

Note that PostgreSQL normalises some names and values: timezone is reported as TimeZone, and "5s" comes back as "5s" while "5000" comes back as "5s" too. Compare against what pg_settings reports rather than what you sent.

TLS deserves an explicit context rather than ssl="require". The string form asks for encryption without verifying the server's identity, which protects against passive eavesdropping and not against an impostor. A context built from the provider's CA bundle verifies both:

import ssl

ssl_context = ssl.create_default_context(cafile="/etc/ssl/certs/rds-global-bundle.pem")
ssl_context.check_hostname = True
ssl_context.verify_mode = ssl.CERT_REQUIRED

Advanced: Per-Service Settings and Measured Tuning

Different entry points in the same codebase want different settings, and copying connect_args into each one is how a timeout ends up missing from the worker. One function with role-specific overrides keeps them together:

import ssl
from typing import Any, Literal

from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

Role = Literal["api", "worker", "report", "migration"]

_BASE_SERVER_SETTINGS = {
    "timezone": "UTC",
    "idle_in_transaction_session_timeout": "30s",
}

_PER_ROLE: dict[Role, dict[str, Any]] = {
    "api":       {"pool_size": 10, "max_overflow": 5,  "statement_timeout": "5s"},
    "worker":    {"pool_size": 4,  "max_overflow": 2,  "statement_timeout": "60s"},
    "report":    {"pool_size": 2,  "max_overflow": 0,  "statement_timeout": "600s"},
    "migration": {"pool_size": 1,  "max_overflow": 0,  "statement_timeout": "0"},
}


def build_engine(url: str, role: Role) -> AsyncEngine:
    config = _PER_ROLE[role]
    return create_async_engine(
        url,
        pool_size=config["pool_size"],
        max_overflow=config["max_overflow"],
        pool_pre_ping=True,
        pool_recycle=1800,
        connect_args={
            "ssl": ssl.create_default_context(),
            "timeout": 10,
            "command_timeout": None if role == "migration" else 120,
            "server_settings": {
                **_BASE_SERVER_SETTINGS,
                "application_name": f"shop-{role}",
                "statement_timeout": config["statement_timeout"],
            },
        },
    )

Three of those choices are deliberate and worth copying. The report role gets a long statement timeout and a tiny pool, so a slow analytical query is allowed to run but cannot occupy the connections the API needs. The migration role disables both timeouts, because a CREATE INDEX legitimately takes longer than any request. And application_name carries the role, so pg_stat_activity grouped by it answers "who is holding these connections" immediately.

Two settings are worth measuring rather than copying.

jit is on by default in PostgreSQL 12 and later, and for short OLTP queries the compilation cost can exceed the execution saving — particularly with many partitions or wide row types. "jit": "off" in server_settings is a one-line experiment: measure p95 for a representative query mix before and after, as in benchmarking asyncpg vs psycopg throughput.

statement_cache_size controls asyncpg's prepared-statement cache. Leaving it at the default is right for a direct connection and wrong behind a transaction-pooling proxy. Do not set it to zero "to be safe": statement preparation is a real saving on a direct connection, and the cost of turning it off is paid on every query.

Finally, pool_recycle interacts with infrastructure rather than with PostgreSQL. Cloud load balancers and NAT gateways commonly drop idle TCP connections after a few minutes, well before any database timeout, so set it below the shortest idle timeout on the path — the reasoning in configuring pool_pre_ping to handle stale connections.

Keeping One Engine Definition for Every Entry Point

The failure this guide is really about is drift: four entry points, four copies of the connection arguments, and only one of them with the setting that matters. A single module that owns engine construction prevents it, and it costs nothing to adopt.

# shop/db.py — the only module that calls create_async_engine
import os

from sqlalchemy.engine import URL
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine

from shop.engine_config import Role, build_engine


def database_url() -> URL:
    return URL.create(
        "postgresql+asyncpg",
        username=os.environ["PGUSER"],
        password=os.environ["PGPASSWORD"],
        host=os.environ["PGHOST"],
        port=int(os.environ.get("PGPORT", 5432)),
        database=os.environ["PGDATABASE"],
    )


def engine_for(role: Role) -> AsyncEngine:
    return build_engine(database_url(), role)


engine = engine_for(os.environ.get("SERVICE_ROLE", "api"))     # type: ignore[arg-type]
Session = async_sessionmaker(engine, expire_on_commit=False)

Two properties follow. A grep -rn "create_async_engine" --include=*.py should find exactly one hit outside tests, which makes the invariant checkable. And adding a setting — a new server parameter, a TLS change, a pool adjustment — is one edit that reaches every entry point, including the ones nobody remembered.

Tests are the legitimate exception, and they should be explicit about it: a test engine wants NullPool or a small pool, no statement timeout that could make a slow test flaky, and its own application_name so a leaked test connection is identifiable in pg_stat_activity.

The last piece is disposing the engine when the process ends, which is not a configuration setting but belongs in the same module, next to the construction. An engine that is never disposed leaves connections to be collected by the garbage collector after the event loop has closed — the warning described in fixing garbage collector non-checked-in connection warnings — and in a process that forks workers, an engine created before the fork is shared in a way that corrupts connections, which disposing async engines on shutdown and in forked workers covers.

Frequently Asked Questions

Related

• Configuring Async Engines and Connection Pools — The parent guide: engine and pool parameters end to end.
• Setting up an async engine from scratch — The engine and lifespan these settings apply to.
• Disposing async engines on shutdown and in forked workers — The other half of engine lifecycle.
• Migrating from psycopg2 to asyncpg — The libpq-to-asyncpg translation table.