Using psycopg 3 async with SQLAlchemy

Use postgresql+psycopg:// with create_async_engine() — psycopg 3 serves synchronous and asynchronous engines from one package, so a codebase migrating gradually keeps one driver, one configuration shape and one exception hierarchy. This guide belongs to choosing between asyncpg and psycopg async drivers.

Quick Answer

psycopg 3 is the only PostgreSQL driver SQLAlchemy can use for both worlds, which is what makes it the pragmatic choice during a long async migration.

Before — two drivers, two configurations:

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine

# Synchronous: psycopg2, libpq-style options.
sync_engine = create_engine(
    "postgresql+psycopg2://shop:secret@db/shop?sslmode=require",
    connect_args={"connect_timeout": 10, "options": "-c statement_timeout=5000"},
)

# Asynchronous: asyncpg, different parameter names entirely.
async_engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    connect_args={"ssl": "require", "timeout": 10,
                  "server_settings": {"statement_timeout": "5000"}},
)

After — one driver, one configuration shape:

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

URL = "postgresql+psycopg://shop:secret@db/shop"
CONNECT_ARGS = {
    "sslmode": "require",
    "connect_timeout": 10,
    "options": "-c statement_timeout=5000",
    "application_name": "orders-api",
}

sync_engine = create_engine(URL, connect_args=CONNECT_ARGS, pool_size=5)
async_engine = create_async_engine(
    URL, connect_args=CONNECT_ARGS, pool_size=10, pool_pre_ping=True
)
Session = async_sessionmaker(async_engine, expire_on_commit=False)

psycopg 3 keeps libpq's connection parameter names, so sslmode, connect_timeout and options work as they did with psycopg2 — which also means an existing connection string needs no translation. Install it as psycopg[binary] (or psycopg[c] where a compiler is available); the pure-Python fallback works but is slower on every statement.

Execution Context & Async Workflow Integration

psycopg 3 is a rewrite of psycopg with an asyncio API alongside the synchronous one, and SQLAlchemy's psycopg dialect drives both. create_engine("postgresql+psycopg://...") uses the synchronous connection class; create_async_engine with the same URL uses AsyncConnection. The SQL generated, the type handling and the exception classes are identical, which is the practical benefit: a service that runs a synchronous admin command and an async web process no longer has two driver behaviours to reason about.

Three configuration details differ from asyncpg and are worth setting deliberately.

Prepared statements are opt-in by threshold. psycopg prepares a statement on the server after it has been executed prepare_threshold times — five by default — and then reuses the plan. That is good for a long-lived connection and wrong behind a transaction-pooling proxy, where the next statement may land on a server connection that never prepared it. Behind PgBouncer in transaction mode, set connect_args={"prepare_threshold": None} to disable preparation entirely. asyncpg has the same problem with a different default and a different fix, both covered in handling asyncpg prepared statement errors with PgBouncer.

Windows needs the selector event loop. psycopg's async implementation does not work with the default ProactorEventLoop, so a Windows development machine needs the policy set before any loop is created:

import asyncio
import sys

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

Timestamps behave like psycopg2, not like asyncpg. psycopg sends timestamps in a form the server interprets, so a naive datetime written to a timestamptz column is interpreted using the session's TimeZone rather than the client's local zone. That is a different failure mode from asyncpg's, and it is why a driver change calls for an audit of datetime handling — the subject of fixing timezone-aware datetime errors with asyncpg. The durable answer is the same for both drivers: timezone-aware columns and aware values everywhere.

Everything else in the application is unchanged. AsyncSession, loader options, stream() for server-side cursors and the pool all behave as they do on asyncpg, because they are SQLAlchemy features rather than driver features.

Resolving Warnings, Errors & Common Mistakes

Checking which implementation is installed takes one line and is worth doing in a startup log, because the pure-Python fallback is easy to get by accident in a container built without wheels:

import psycopg

print(psycopg.pq.__impl__)   # "binary", "c", or "python"

The portable way to handle errors is the SQLSTATE code, which comes from PostgreSQL and is therefore identical across psycopg2, psycopg 3 and asyncpg:

from sqlalchemy.exc import DBAPIError

UNIQUE_VIOLATION = "23505"
FOREIGN_KEY_VIOLATION = "23503"
DEADLOCK_DETECTED = "40P01"
SERIALIZATION_FAILURE = "40001"


def sqlstate(exc: DBAPIError) -> str | None:
    for candidate in (exc.orig, getattr(exc.orig, "__cause__", None)):
        code = getattr(candidate, "sqlstate", None) or getattr(candidate, "pgcode", None)
        if code:
            return code
    return None

Adopting that helper before changing drivers is what makes the change reversible: the error handling stops depending on which driver is underneath, so switching back and forth is a URL change rather than a code change. The rest of the driver-independent work is listed in migrating from psycopg2 to asyncpg, and it applies equally to a move to psycopg 3.

Advanced: Pipeline Mode, COPY and Server-Side Cursors

Three psycopg 3 capabilities are worth knowing because they change what bulk work costs.

Pipeline mode sends several statements without waiting for each result, which removes a round trip per statement. SQLAlchemy's psycopg dialect uses it for executemany, so a Core insert with a list of dictionaries is dramatically faster than the same call on psycopg2 — often close to COPY for moderate batches:

from sqlalchemy import insert

from shop.models import PageView


async def record_views(session, rows: list[dict]) -> None:
    await session.execute(insert(PageView), rows)     # pipelined executemany
    await session.commit()

COPY is available from the cursor in both the synchronous and asynchronous APIs, reached through the session's connection:

async def load_page_views(session, rows: list[tuple]) -> None:
    connection = await session.connection()
    raw = await connection.get_raw_connection()
    async with raw.driver_connection.cursor() as cursor:
        async with cursor.copy(
            "COPY page_views (viewed_at, customer_id, path, duration_ms) FROM STDIN"
        ) as copy:
            for row in rows:
                await copy.write_row(row)
    await session.commit()

Because it runs on the session's own connection, the load is inside the session's transaction — the same property the asyncpg version has, as described in loading rows with Postgres COPY through asyncpg. psycopg's copy() accepts text or binary format and takes rows as tuples, with adaptation handled by the driver, which makes it more forgiving about types than asyncpg's binary copy_records_to_table.

Server-side cursors back AsyncSession.stream(), so large reads work the same way they do on asyncpg:

from sqlalchemy import select

from shop.models import Order

result = await session.stream_scalars(
    select(Order).order_by(Order.id).execution_options(yield_per=1_000)
)
async with result:
    async for order in result:
        ...

What psycopg 3 does not match is asyncpg's raw throughput on large result sets: asyncpg's protocol implementation and binary decoding are faster, and the gap widens as row counts grow. That is the trade-off in one sentence — psycopg 3 for one driver across both worlds and richer features, asyncpg for maximum read throughput — and it is worth deciding with a measurement from your own workload, as benchmarking asyncpg vs psycopg throughput sets out.

Running Both Drivers From One Configuration

During a migration it is useful to be able to switch drivers without editing code, so the choice can be made per environment and reverted instantly if something misbehaves. Because SQLAlchemy hides the driver, that is a matter of building the URL and the connect arguments from one place.

# shop/db.py
import os
import sys
from typing import Any

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

DRIVER = os.environ.get("PG_ASYNC_DRIVER", "psycopg")      # or "asyncpg"


def _connect_args(driver: str) -> dict[str, Any]:
    if driver == "asyncpg":
        return {
            "ssl": "require",
            "timeout": 10,
            "server_settings": {
                "application_name": "orders-api",
                "statement_timeout": "5000",
            },
        }
    return {
        "sslmode": "require",
        "connect_timeout": 10,
        "application_name": "orders-api",
        "options": "-c statement_timeout=5000",
    }


def build_url(driver: str) -> URL:
    return URL.create(
        drivername=f"postgresql+{driver}",
        username=os.environ["PGUSER"],
        password=os.environ["PGPASSWORD"],     # no escaping needed: URL.create handles it
        host=os.environ["PGHOST"],
        port=int(os.environ.get("PGPORT", 5432)),
        database=os.environ["PGDATABASE"],
    )


if DRIVER == "psycopg" and sys.platform == "win32":
    import asyncio

    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

engine = create_async_engine(
    build_url(DRIVER),
    connect_args=_connect_args(DRIVER),
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)
Session = async_sessionmaker(engine, expire_on_commit=False)

URL.create() is worth using in preference to a formatted string: it escapes the password correctly, so a password containing @, / or % needs no attention, and it keeps the credentials out of any log line that prints the URL, because SQLAlchemy renders URL objects with the password masked.

With that in place, three things become easy. The integration suite can run under both drivers in CI, which surfaces behavioural differences early — type conversions, error message text, timestamp handling. A canary deployment can run one driver while the rest of the fleet runs the other, with the same image. And a rollback is an environment variable.

The remaining driver-specific code should be small enough to list: the connect arguments above, and anything that reaches driver_connection for COPY. Keeping those in one module, rather than at each call site, is what keeps the switch to a single file — and is the same discipline that makes the eventual choice, whichever way it goes, cheap to commit to.

Frequently Asked Questions

Related

• Choosing Between asyncpg and psycopg Async Drivers — The parent guide: the two drivers compared.
• Benchmarking asyncpg vs psycopg throughput — Measuring the difference on your own workload.
• Migrating from psycopg2 to asyncpg — The driver-independent work that makes any switch reversible.
• Handling asyncpg prepared statement errors with PgBouncer — The same proxy problem, on the other driver.