Using IAM database authentication with async engines

Listen for do_connect on engine.sync_engine and set cparams["password"] to a freshly generated IAM token each time the pool opens a connection — tokens expire in minutes but are only checked at login, so pooled connections keep working while new ones always get a valid credential. This guide belongs to tuning connection pools for cloud databases.

Quick Answer

The tempting version generates a token at startup and puts it in the URL. It works until the first connection opened after the token expires.

Before — a token baked into the URL at startup:

import boto3
from sqlalchemy.ext.asyncio import create_async_engine

rds = boto3.client("rds", region_name="eu-west-1")
token = rds.generate_db_auth_token(
    DBHostname="shop.cluster-abc.eu-west-1.rds.amazonaws.com", Port=5432,
    DBUsername="orders_api", Region="eu-west-1",
)
engine = create_async_engine(
    f"postgresql+asyncpg://orders_api:{token}@shop.cluster-abc.eu-west-1.rds.amazonaws.com/shop"
)
# Sixteen minutes later, when the pool opens another connection:
# asyncpg.exceptions.InvalidPasswordError: PAM authentication failed for user "orders_api"

After — a token per new connection, cached for most of its life:

import ssl
import time

import boto3
from sqlalchemy import event
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

HOST = "shop.cluster-abc.eu-west-1.rds.amazonaws.com"
PORT = 5432
USER = "orders_api"
REGION = "eu-west-1"
TOKEN_TTL_S = 600   # refresh well inside the 15-minute validity

_rds = boto3.client("rds", region_name=REGION)
_cached: tuple[float, str] | None = None


def _iam_token() -> str:
    global _cached
    now = time.monotonic()
    if _cached is None or now - _cached[0] > TOKEN_TTL_S:
        token = _rds.generate_db_auth_token(
            DBHostname=HOST, Port=PORT, DBUsername=USER, Region=REGION
        )
        _cached = (now, token)
    return _cached[1]


engine = create_async_engine(
    f"postgresql+asyncpg://{USER}@{HOST}:{PORT}/shop",
    connect_args={"ssl": ssl.create_default_context()},
    pool_size=10,
    pool_pre_ping=True,
    pool_recycle=3600,
)


@event.listens_for(engine.sync_engine, "do_connect")
def _provide_token(dialect, conn_rec, cargs, cparams) -> None:
    cparams["password"] = _iam_token()


Session = async_sessionmaker(engine, expire_on_commit=False)

Download the RDS certificate bundle and load it into the SSLContext with load_verify_locations() if the default trust store does not include Amazon's roots in your image.

Execution Context & Async Workflow Integration

IAM database authentication replaces a long-lived password with a short-lived signed token. On AWS, generate_db_auth_token() produces a SigV4-signed string — computed locally from the process's AWS credentials, with no network call — that the database accepts as the password for a user granted the rds_iam role, for fifteen minutes. The same model appears on Azure, where an Entra ID access token is the password, and on Google Cloud SQL, where the connector library handles tokens for you.

The key property is that tokens authenticate logins. PostgreSQL checks the credential once, while establishing the connection, and never again. A pooled connection opened with a token that has since expired keeps working for as long as the connection lives. What fails is the next new connection — from pool growth into overflow, pool_recycle replacing an old connection, pool_pre_ping replacing a dead one, or a failover — if it presents an expired token.

do_connect is the hook that fits that exactly. SQLAlchemy fires it immediately before the driver's connect() call, with the positional and keyword arguments it is about to pass. Changing cparams["password"] changes what asyncpg receives for that one connection. The listener is registered on engine.sync_engine, because pool and connection events live on the synchronous engine the async engine wraps, and it runs synchronously inside SQLAlchemy's connection logic.

Because the listener is synchronous and runs on the event loop thread, what it does matters. AWS token generation is local signing and takes well under a millisecond, so calling it on every connect would be acceptable; caching it for ten minutes makes it essentially free. Azure's DefaultAzureCredential.get_token() can make a network call to an identity endpoint, which would block the loop — cache the token and refresh it before expiry, ideally from a background task, so the listener only ever reads a cached value.

IAM authentication requires TLS; a connection attempt without it is rejected by pg_hba.conf rules before the token is considered. Set ssl in connect_args with a verifying context. Pool settings are otherwise unaffected, and the guidance in setting pool_size and max_overflow for AWS RDS applies unchanged.

Resolving Warnings, Errors & Common Mistakes

Two of these are worth expanding.

The hostname row catches setups with several endpoints. The token is a signature over the hostname, port, user and region. A token generated for the cluster writer endpoint is not valid when connecting to the reader endpoint, a custom endpoint or an RDS Proxy endpoint, even for the same user. Engines for read replicas each need their own listener, signing for their own host.

The immediate failure is usually permissions rather than code. Three things must all be true: IAM authentication is enabled on the instance or cluster; the database user exists and has been granted rds_iam (and, once it has, can no longer log in with a password); and the calling principal's IAM policy allows rds-db:connect on arn:aws:rds-db:<region>:<account>:dbuser:<resource-id>/<user>. The resource ID in that ARN is the cluster's or instance's DbiResourceId, not its name, which is the most common policy mistake.

Advanced: Cloud SQL With async_creator, and Azure Tokens

Google's Cloud SQL Python Connector does more than generate tokens: it fetches ephemeral client certificates, establishes the TLS connection to the instance and handles IAM login. It therefore replaces the whole connection step rather than supplying a password, and SQLAlchemy's async_creator parameter is the place to plug it in:

from google.cloud.sql.connector import Connector, create_async_connector
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

INSTANCE = "shop-prod:europe-west1:shop-pg"


async def build_engine() -> tuple[AsyncEngine, Connector]:
    connector = await create_async_connector()

    async def getconn():
        return await connector.connect_async(
            INSTANCE,
            "asyncpg",
            user="orders-api@shop-prod.iam",   # service account, without .gserviceaccount.com
            db="shop",
            enable_iam_auth=True,
        )

    engine = create_async_engine(
        "postgresql+asyncpg://",
        async_creator=getconn,
        pool_size=5,
        pool_pre_ping=True,
    )
    return engine, connector


# In the application lifespan:
#   engine, connector = await build_engine()
#   ...
#   await engine.dispose()
#   await connector.close_async()

async_creator is called whenever the pool needs a connection, and because it is a coroutine, the connector's network calls do not block the loop. The URL carries only the dialect and driver; everything else comes from the connector. Create the connector inside the running loop, as above, rather than at import time.

Azure Database for PostgreSQL uses Entra ID access tokens as passwords, valid for roughly an hour. The token fetch can involve a network call, so fetch and refresh it asynchronously and let do_connect read the cached value:

import asyncio
import time

from azure.identity.aio import DefaultAzureCredential
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine

SCOPE = "https://ossrdbms-aad.database.windows.net/.default"


class AzureTokenCache:
    def __init__(self) -> None:
        self._credential = DefaultAzureCredential()
        self.token: str | None = None

    async def refresh_forever(self) -> None:
        while True:
            access = await self._credential.get_token(SCOPE)
            self.token = access.token
            # Refresh five minutes before expiry.
            await asyncio.sleep(max(60, access.expires_on - time.time() - 300))


def install_azure_auth(engine: AsyncEngine, cache: AzureTokenCache) -> None:
    @event.listens_for(engine.sync_engine, "do_connect")
    def _provide_token(dialect, conn_rec, cargs, cparams):
        if cache.token is None:
            raise RuntimeError("Azure token not yet available")
        cparams["password"] = cache.token

Start refresh_forever() as a task in the lifespan and await the first token before the application accepts traffic. The pattern generalises to any provider whose token fetch is slow: refresh in the background, read synchronously at connect time.

IAM Authentication With Proxies and Migrations

Two parts of a deployment connect to the database outside the request path, and both need the same treatment: connection proxies and migration runs.

RDS Proxy supports IAM authentication on its client side, so the application signs a token for the proxy endpoint and the proxy authenticates to the database using credentials from Secrets Manager. The do_connect listener is identical; only the hostname it signs for changes. Because the proxy multiplexes client connections, tokens are validated when the application opens a connection to the proxy — which, with the NullPool pattern used in serverless handlers, is every invocation. Caching the token matters more there, since it removes signing from the per-invocation path entirely.

PgBouncer in front of RDS cannot pass IAM tokens through in the usual configurations: it authenticates clients itself and connects to the database with its own credentials. Applications behind it typically use a password known to PgBouncer while PgBouncer's own upstream connection uses IAM or a managed secret. Check what your proxy supports before designing around end-to-end IAM.

Migrations run with a different engine, usually from a CI job or release step, and need tokens too. An async env.py can install the same listener on the engine it creates, or reuse the application's engine through the shared-connection pattern in running Alembic migrations programmatically from async code:

# alembic/env.py (excerpt)
from sqlalchemy import event, pool
from sqlalchemy.ext.asyncio import async_engine_from_config

from shop.db_auth import iam_token, ssl_context


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
        connect_args={"ssl": ssl_context()},
    )

    @event.listens_for(connectable.sync_engine, "do_connect")
    def _provide_token(dialect, conn_rec, cargs, cparams):
        cparams["password"] = iam_token()

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

Migration users usually need more privileges than the application user — creating tables, altering types — so they are typically a separate database role with its own rds_iam grant and its own IAM policy statement. Keeping them separate means the application's credentials cannot alter the schema, and the migration job's credentials exist only for the minutes the job runs.

Frequently Asked Questions

Related

• Tuning Connection Pools for Cloud Databases — The parent guide: pool sizing for managed databases.
• Setting pool_size and max_overflow for AWS RDS — Pool settings that pair with IAM-authenticated engines.
• Using NullPool for serverless and AWS Lambda — Per-invocation connections, where token caching matters most.
• Configuring pool_pre_ping to handle stale connections — Replacement connections that each need a fresh token.