Migrating from psycopg2 to asyncpg

Change the URL to postgresql+asyncpg://, translate sslmode, connect_timeout and options into ssl, timeout and server_settings, replace every direct psycopg2 call with a SQLAlchemy construct, and catch IntegrityError by SQLSTATE rather than psycopg2 exception classes — then audit datetimes, because asyncpg's binary protocol is stricter. This guide belongs to choosing between asyncpg and psycopg async drivers.

Quick Answer

Most of a SQLAlchemy application does not care which driver is underneath. The parts that do are the connection options, any code that touched psycopg2 directly, and error handling that named psycopg2 exceptions.

Before — a psycopg2 engine with driver-specific options and code:

import psycopg2.errors
from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError

engine = create_engine(
    "postgresql+psycopg2://shop:secret@db/shop?sslmode=require&connect_timeout=10",
    connect_args={"options": "-c statement_timeout=5000", "application_name": "orders-api"},
    pool_size=10,
)


def create_customer(session, email: str):
    try:
        session.connection().exec_driver_sql(
            "INSERT INTO customers (email) VALUES (%(email)s)", {"email": email}
        )
        session.commit()
    except IntegrityError as exc:
        if isinstance(exc.orig, psycopg2.errors.UniqueViolation):
            raise EmailTaken(email) from exc
        raise

After — asyncpg, with options translated and portable SQL and error handling:

import ssl

from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

UNIQUE_VIOLATION = "23505"

engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    connect_args={
        "ssl": ssl.create_default_context(),
        "timeout": 10,
        "server_settings": {"statement_timeout": "5000", "application_name": "orders-api"},
    },
    pool_size=10,
    pool_pre_ping=True,
)


def sqlstate(exc: IntegrityError) -> 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


async def create_customer(session: AsyncSession, email: str) -> None:
    try:
        await session.execute(
            text("INSERT INTO customers (email) VALUES (:email)"), {"email": email}
        )
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        if sqlstate(exc) == UNIQUE_VIOLATION:
            raise EmailTaken(email) from exc
        raise

:email placeholders in text() are SQLAlchemy's own and are translated for whichever driver is in use, so that SQL is now portable. The SQLSTATE check works across psycopg2, psycopg 3 and asyncpg, which makes it worth adopting before the switch.

Execution Context & Async Workflow Integration

psycopg2 and asyncpg differ at every level below SQLAlchemy, and knowing where helps predict what will break.

Connection parameters. psycopg2 passes connection options to libpq, the C client library, which understands sslmode, connect_timeout, options and application_name as part of its connection string. asyncpg is a pure-Python-and-Cython client with no libpq, and its connect() has its own parameters: ssl takes True, a mode string, or an SSLContext; timeout is the connect timeout; and run-time settings go in a server_settings dictionary that asyncpg sends at startup. SQLAlchemy's asyncpg dialect translates a few URL query parameters, including ssl, but passes anything it does not recognise straight to asyncpg.connect(), which is where TypeError: connect() got an unexpected keyword argument 'sslmode' comes from.

Protocol. psycopg2 sends parameters as text, interpolated client-side into the query string, and lets the server parse values. asyncpg uses PostgreSQL's extended protocol with server-side prepared statements and binary encoding, which is faster and stricter: a value must match its parameter's type on the client. Most mismatches never appear because SQLAlchemy types the parameters, but hand-written text() SQL with ambiguous parameters, and datetimes, are exceptions — the latter covered in fixing timezone-aware datetime errors with asyncpg. Prepared statements also interact with PgBouncer in transaction mode, which handling asyncpg prepared statement errors with PgBouncer covers.

Concurrency model. psycopg2 is synchronous: concurrency comes from threads or processes, and a blocked thread cannot run another query. asyncpg is asyncio-only, and SQLAlchemy exposes it only through create_async_engine and AsyncSession. That is the largest change by far, because it reaches past the database layer: every function that touches a session becomes async def, every caller of those functions awaits them, and implicit lazy loading of relationships, which psycopg2-backed code relied on without noticing, raises MissingGreenlet and has to become explicit eager loading.

Errors. SQLAlchemy wraps both drivers' exceptions in the same hierarchy — IntegrityError, OperationalError, DBAPIError — so code that catches those is portable. Code that inspects exc.orig for psycopg2 classes such as psycopg2.errors.UniqueViolation is not, because under asyncpg exc.orig is SQLAlchemy's adapter around asyncpg.exceptions.UniqueViolationError. The five-character SQLSTATE code is the common ground.

Resolving Warnings, Errors & Common Mistakes

could not determine data type of parameter $1 is the one that surprises teams with well-tested SQL. psycopg2 interpolates the value as text before the server sees the query, so SELECT :value IS NULL works. asyncpg asks the server to prepare the statement first, and PostgreSQL cannot infer a type for a parameter that appears only in IS NULL or in a COALESCE with no typed argument. Give it one:

from sqlalchemy import String, bindparam, text

stmt = text(
    "SELECT id FROM customers WHERE (:email IS NULL OR email = :email)"
).bindparams(bindparam("email", type_=String))

MissingGreenlet is the error that takes the most time, because the fix is structural rather than local. Every relationship access that psycopg2-era code performed lazily has to become a loader option on the query that loads the parent — the patterns are in fixing GreenletSpawnError in async SQLAlchemy workflows. Setting lazy="raise" on relationships before the switch, while still on psycopg2, turns every hidden lazy load into an immediate, synchronous error you can fix one at a time.

Advanced: Replacing psycopg2 Extras and Rolling Out Gradually

psycopg2's extras module and cursor methods are the main source of driver-specific code, and each has a direct replacement.

execute_values() for batch inserts becomes a Core insert with a list of dictionaries, which SQLAlchemy sends through asyncpg's efficient executemany:

from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import PageView


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

copy_expert() and copy_from() become asyncpg's COPY methods on the underlying connection, which are faster than their psycopg2 counterparts — see loading rows with Postgres COPY through asyncpg. RealDictCursor becomes result.mappings(). register_adapter and register_type for custom types become a TypeDecorator, which works with every driver.

The rollout itself is safest as a sequence of independently shippable steps, most of which happen while still on psycopg2:

1. Remove direct driver usage. grep -rn "psycopg2" --include=*.py should find nothing outside configuration. Replace cursors, extras and exception classes as above.
2. Make loading explicit. Add lazy="raise" to relationships and fix what breaks, so nothing depends on implicit I/O.
3. Run both engines. Create the async engine alongside the sync one, pointing at the same database with a separate, smaller pool, and move entry points over one at a time — a background worker first, then read-only endpoints, then writes.
4. Audit behaviour differences. Datetimes, untyped parameters and SQLSTATE handling, each with a test.
5. Remove the sync engine once nothing uses it, keeping Alembic on whichever driver its env.py is written for.

If step three looks too large — a big synchronous codebase that cannot become async def quickly — psycopg 3 is worth considering instead. It supports both synchronous and asyncio use with one driver, so a service can move its connection layer first and adopt async incrementally. The trade-offs are laid out in the parent guide on choosing between asyncpg and psycopg async drivers.

Verifying the Switch in Tests and Staging

A driver change is invisible when it works and scattered when it does not, so verification needs to exercise the specific differences rather than just run the existing suite.

Run the suite against both drivers during the transition. Parametrise the engine fixture so the same tests run on psycopg2 and on asyncpg; a test that passes on one and fails on the other points directly at a behavioural difference. Tests that only exercise SQLAlchemy constructs will pass on both, which is itself useful confirmation.

Add targeted tests for the known differences. Four are worth a test each, because they fail silently or only under specific data:

import datetime as dt

import pytest
from sqlalchemy import String, bindparam, text
from sqlalchemy.exc import IntegrityError

from shop.errors import sqlstate


@pytest.mark.asyncio
async def test_unique_violation_is_detected_by_sqlstate(session, customer_factory):
    await customer_factory(email="a@example.com")
    with pytest.raises(IntegrityError) as caught:
        await session.execute(
            text("INSERT INTO customers (email) VALUES (:email)"), {"email": "a@example.com"}
        )
    assert sqlstate(caught.value) == "23505"


@pytest.mark.asyncio
async def test_timestamps_round_trip_as_aware_utc(session, order_factory):
    placed = dt.datetime(2026, 9, 17, 9, 0, tzinfo=dt.UTC)
    order = await order_factory(placed_at=placed)
    await session.refresh(order)
    assert order.placed_at == placed and order.placed_at.tzinfo is not None


@pytest.mark.asyncio
async def test_optional_filter_parameter_is_typed(session):
    rows = await session.execute(
        text("SELECT 1 WHERE (:email IS NULL OR :email = 'x')").bindparams(
            bindparam("email", type_=String)
        ),
        {"email": None},
    )
    assert rows.scalar() == 1

The fourth is the connection options: a startup check that reads back SHOW statement_timeout and current_setting('application_name') and fails if they are not what the configuration intended, because a dropped server_settings entry produces no error at all.

Watch staging for what tests cannot see. Pool behaviour changes with the concurrency model: an async service holds connections across await points and serves many more requests per process, so the pool size that suited a threaded service is often too small or too large. Compare pool.checkedout() and database backend counts in staging under realistic load before production, using the sizing approach in setting up an asyncpg connection pool for high concurrency.

Frequently Asked Questions

Related

• Choosing Between asyncpg and psycopg Async Drivers — The parent guide: the two drivers compared.
• Using SQLAlchemy async with Celery task workers — Running the async engine inside synchronous workers.
• Fixing timezone-aware datetime errors with asyncpg — The most common behaviour change after switching.
• Setting up an async engine from scratch — The engine and lifecycle the migration ends with.