Selecting Async Drivers for SQLite, MySQL, and Postgres

Choosing the right async database driver is one of the first architecture decisions you make when building an async Python service, and the answer is backend-specific. This guide — part of the Async Engines, Dialects, and Connection Pooling reference — maps each major backend to its mature async driver options, explains the URL scheme differences (postgresql+asyncpg://, mysql+aiomysql://, sqlite+aiosqlite://), and gives you the practical knowledge to pick the right driver for production, staging, and test environments.

Concept & Execution Model

SQLAlchemy 2.0 introduced a unified async surface through create_async_engine() and AsyncSession. The engine accepts a dialect-qualified URL where the prefix before + identifies the backend (postgresql, mysql, sqlite) and the suffix identifies the driver. SQLAlchemy's dialect layer translates ORM operations into driver-specific wire calls transparently — but the driver's execution model, pooling behaviour, and feature set differ significantly.

The three async drivers you will encounter in everyday Python async development are:

• asyncpg and psycopg3 for PostgreSQL
• aiomysql and asyncmy for MySQL and MariaDB
• aiosqlite for SQLite

Each driver wraps a native async or asyncio-compatible interface for its respective database protocol. None of them share code; they have distinct connection lifecycle semantics, distinct pool compatibility characteristics, and distinct behaviours when confronted with SQLAlchemy's autobegin/autoflush machinery.

The matrix below maps each backend and environment to the driver and pool class this guide recommends — an at-a-glance decision map before the sections that follow explain the reasoning.

The relationship between URL scheme, driver package, and SQLAlchemy dialect string:

# PostgreSQL — asyncpg (native async, binary protocol)
"postgresql+asyncpg://user:password@host:5432/dbname"

# PostgreSQL — psycopg3 (libpq-backed async)
"postgresql+psycopg://user:password@host:5432/dbname"

# MySQL — aiomysql (asyncio wrapper over pymysql)
"mysql+aiomysql://user:password@host:3306/dbname"

# MySQL — asyncmy (pure-Python async MySQL)
"mysql+asyncmy://user:password@host:3306/dbname"

# SQLite — aiosqlite (asyncio wrapper over sqlite3)
"sqlite+aiosqlite:///path/to/local.db"

# SQLite — in-memory (tests/CI)
"sqlite+aiosqlite://"

SQLAlchemy builds an AsyncEngine from any of these URLs with identical Python code:

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

engine = create_async_engine(
    "postgresql+asyncpg://app:secret@db:5432/production",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=1800,
    echo=False,
)

AsyncSessionFactory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Swapping the URL is all that changes between backends from the application's perspective. The sections below examine what changes underneath.

Query Construction & Async Execution Patterns

SQLAlchemy 2.0's select() / execute() / scalars() pipeline is dialect-agnostic. The same query code runs against Postgres, MySQL, or SQLite without modification for standard operations. Driver differences surface in type-specific behaviour, auto-increment semantics, and SQL dialect extensions.

The 2.0 query surface is identical between the synchronous and asynchronous execution paths — the only difference is the await and the async session/engine constructors. Everything else, including the select() statement itself, is shared verbatim:

# Sync (psycopg2 / pymysql / sqlite3) — SQLAlchemy 2.0 style
from sqlalchemy import select
from sqlalchemy.orm import Session

def get_active_users_sync(session: Session, tenant_id: int) -> list[User]:
    stmt = select(User).where(User.tenant_id == tenant_id).where(User.is_active.is_(True))
    return list(session.execute(stmt).scalars().all())

# Async (asyncpg / asyncmy / aiosqlite) — same statement, awaited execution
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_active_users_async(session: AsyncSession, tenant_id: int) -> list[User]:
    stmt = select(User).where(User.tenant_id == tenant_id).where(User.is_active.is_(True))
    result = await session.execute(stmt)
    return list(result.scalars().all())

The choice of driver, not the query code, is what changes between backends — which is exactly why driver selection is the decision worth getting right up front.

Standard async query pattern

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, nullable=False)
    tenant_id: Mapped[int] = mapped_column(nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)


async def get_active_users(session: AsyncSession, tenant_id: int) -> list[User]:
    stmt = (
        select(User)
        .where(User.tenant_id == tenant_id)
        .where(User.is_active.is_(True))
        .order_by(User.email)
    )
    result = await session.execute(stmt)
    return result.scalars().all()


async def main(db_url: str) -> None:
    engine = create_async_engine(db_url, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with factory() as session:
        users = await get_active_users(session, tenant_id=1)
        print(f"Found {len(users)} active users")
    await engine.dispose()

# Works with any of the URL schemes above — swap the string, keep the logic.
asyncio.run(main("postgresql+asyncpg://app:secret@localhost/mydb"))

Backend-specific type behaviour

The most common source of cross-backend surprises involves types that SQLAlchemy maps differently per dialect:

When writing portable code that runs against multiple backends (Postgres in production, SQLite in tests), avoid ARRAY, JSONB, and Postgres-specific column types in your ORM models. Use JSON from sqlalchemy (not sqlalchemy.dialects.postgresql) and Boolean rather than SmallInteger for boolean columns.

State Management & Session Boundaries

Session semantics are the same across all backends at the SQLAlchemy level — async with session.begin(): always delimits a transaction — but the underlying connection lifecycle differs in ways that affect production configuration.

Connection pooling compatibility by backend

SQLAlchemy's QueuePool (the default) works with asyncpg, psycopg3, and aiomysql/asyncmy. For aiosqlite (SQLite), you must use StaticPool for in-memory databases (where a new connection equals a new blank database) and should use NullPool or StaticPool for file-based databases in test contexts. SQLite's threading model predates async and its C library enforces per-thread connection ownership unless check_same_thread=False is explicitly set — aiosqlite handles this automatically, but the pool must be configured to hand back the same underlying connection.

from sqlalchemy.pool import StaticPool

# In-memory SQLite: StaticPool + shared connection across sessions
test_engine = create_async_engine(
    "sqlite+aiosqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

# File-based SQLite in tests: NullPool to avoid thread-crossing
file_engine = create_async_engine(
    "sqlite+aiosqlite:///./test.db",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

# Production Postgres: QueuePool (default) with explicit sizing
prod_engine = create_async_engine(
    "postgresql+asyncpg://app:secret@db:5432/production",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=1800,
)

The full walkthrough for in-memory SQLite in async test fixtures — including the StaticPool gotcha and pytest-asyncio session scope — is covered in the guide on using aiosqlite for async tests and local development.

Transaction autocommit differences

MySQL's default isolation level is REPEATABLE READ, while PostgreSQL defaults to READ COMMITTED. SQLite has no concept of server-side isolation — isolation is entirely a per-connection concern. When targeting MySQL with aiomysql or asyncmy, be aware that SELECT inside a transaction takes a consistent snapshot at the transaction's start, which can surprise developers accustomed to Postgres READ COMMITTED semantics where a re-query mid-transaction sees committed data from other sessions.

# Explicitly set isolation level per engine — works for all async drivers
engine_mysql = create_async_engine(
    "mysql+aiomysql://app:secret@mysql:3306/production",
    isolation_level="READ COMMITTED",   # override MySQL default
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
    pool_recycle=3600,   # MySQL drops idle connections at 8h by default
)

Advanced Driver-Specific Patterns

PostgreSQL: asyncpg vs psycopg3 decision boundary

For PostgreSQL, you have two strong driver options. asyncpg offers native-async execution with binary-protocol transfer for all supported types (zero string-serialisation overhead for INTEGER, UUID, JSONB, TIMESTAMP), making it the highest-throughput choice for OLTP workloads. psycopg3 wraps libpq with asyncio events, supports PostgreSQL's COPY protocol natively through cursor.copy(), and integrates more cleanly with PgBouncer's transaction-pooling mode because its prepare_threshold=None setting disables server-side prepared statements without requiring the connection state surgery needed for asyncpg.

The detailed comparison — including prepared-statement cache sizes, PgBouncer statement_cache_size=0 requirements, and COPY throughput benchmarks — is in the dedicated guide on choosing between asyncpg and psycopg3.

Rule of thumb:

• asyncpg for latency-sensitive OLTP APIs, pure-Python deployments (no libpq system dependency), and read-heavy workloads
• psycopg3 for bulk-ingest pipelines using native COPY, PgBouncer transaction-pool compatibility without configuration surgery, and teams migrating from psycopg2

Installation:

# asyncpg
pip install asyncpg sqlalchemy[asyncio]

# psycopg3
pip install "psycopg[async]" sqlalchemy[asyncio]
# Optional binary extension for better performance:
pip install "psycopg[binary,async]"

MySQL/MariaDB: aiomysql vs asyncmy

For MySQL and MariaDB, the two async driver options are aiomysql and asyncmy. Both use the MySQL binary protocol over asyncio but have distinct maintenance and feature trajectories.

aiomysql is the older, more widely deployed option. It is an asyncio port of PyMySQL and has broad compatibility across MySQL 5.7, MySQL 8.x, and MariaDB. Most existing SQLAlchemy async MySQL documentation targets aiomysql. Its pool_recycle configuration is important because MySQL's wait_timeout (default 8 hours, often 600 seconds on managed cloud instances) silently drops idle connections.

asyncmy is a newer, pure-Python async MySQL driver with a cleaner internal architecture and faster development velocity as of 2024–2025. It supports MySQL 8.x features like caching_sha2_password authentication (which aiomysql had early-adoption issues with) and MariaDB 10.x column-level encoding changes.

# aiomysql — battle-tested, widest compatibility
engine_aiomysql = create_async_engine(
    "mysql+aiomysql://app:secret@mysql:3306/production?charset=utf8mb4",
    pool_size=15,
    max_overflow=5,
    pool_pre_ping=True,
    pool_recycle=3600,          # recycle before MySQL wait_timeout
    connect_args={
        "connect_timeout": 10,
        "autocommit": False,    # SQLAlchemy manages transactions
    },
)

# asyncmy — modern driver, prefer for MySQL 8.x / MariaDB 10.6+
engine_asyncmy = create_async_engine(
    "mysql+asyncmy://app:secret@mysql:3306/production?charset=utf8mb4",
    pool_size=15,
    max_overflow=5,
    pool_pre_ping=True,
    pool_recycle=3600,
)

MariaDB note: MariaDB 10.6+ introduced a strict latin1 vs utf8mb4 default collation change that breaks implicit character set assumptions. Always specify ?charset=utf8mb4 in your connection URL for both drivers.

Installation:

# aiomysql
pip install aiomysql sqlalchemy[asyncio]

# asyncmy
pip install asyncmy sqlalchemy[asyncio]

Managed MySQL cloud considerations: AWS RDS MySQL and Google Cloud SQL both set wait_timeout to values well below MySQL's built-in 8-hour default. RDS MySQL defaults to 28800 seconds but many DBA teams lower it to 600–1800 seconds for cost reasons. Always verify the actual wait_timeout with SHOW VARIABLES LIKE 'wait_timeout'; and set pool_recycle to at most half that value. On Cloud SQL, the proxy already handles reconnection to some extent, but SQLAlchemy's pool is unaware of the proxy's behaviour — pool_pre_ping=True remains essential.

SQLite: aiosqlite for local and test environments

aiosqlite wraps Python's built-in sqlite3 module with asyncio, using a background thread per connection to avoid blocking the event loop (SQLite is not natively async — its C library is single-threaded). This means aiosqlite is fundamentally a synchronous executor wrapped in async sugar; it never achieves true non-blocking I/O at the OS level. That is fine for local development and test workloads where the database is in-memory or on local disk and I/O latency is measured in microseconds, not milliseconds.

For production workloads, SQLite with aiosqlite is appropriate only for:

• Single-process embedded applications where the database file and the application process are co-located
• Low-write, read-mostly workloads (SQLite's write lock is database-wide, not row-level)
• Desktop or CLI tools where no network database is available
• Edge deployments on Cloudflare Workers or similar runtimes that expose an SQLite-compatible API (e.g., D1 via a compatibility shim)

Installation:

pip install aiosqlite sqlalchemy[asyncio]

import asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.pool import StaticPool


async def demo_sqlite() -> None:
    """
    In-memory SQLite with StaticPool: all sessions share one connection,
    so schema created in one session is visible in subsequent sessions.
    """
    engine = create_async_engine(
        "sqlite+aiosqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
        echo=True,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async with factory() as session:
        async with session.begin():
            session.add(User(email="alice@example.com", tenant_id=1))

    async with factory() as session:
        result = await session.execute(select(User).where(User.tenant_id == 1))
        users = result.scalars().all()
        print(f"Retrieved {len(users)} users from in-memory SQLite")

    await engine.dispose()


asyncio.run(demo_sqlite())

Hybrid Architectures & Migration Strategies

Running Postgres in production with SQLite in tests

The most common multi-backend configuration is Postgres for production and SQLite for fast, dependency-free local tests. This is a legitimate and widely used architecture — but it requires discipline about which SQLAlchemy constructs you use.

Safe for cross-backend use:

• select(), insert(), update(), delete() with standard column types
• func.count(), func.sum(), func.coalesce()
• ORM relationships (selectinload, joinedload, subqueryload)
• JSON type from sqlalchemy (not sqlalchemy.dialects.postgresql)

Postgres-only (avoid in portable models):

• ARRAY, TSVECTOR, JSONB, HSTORE, INET
• func.array_agg(), func.json_build_object()
• ON CONFLICT DO UPDATE (insert().on_conflict_do_update())
• Window functions using Postgres-specific frame syntax

# conftest.py — pytest setup that works with both backends
import asyncio
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.pool import StaticPool


DATABASE_URL = "sqlite+aiosqlite://"   # override per environment


@pytest_asyncio.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest_asyncio.fixture(scope="session")
async def engine():
    _engine = create_async_engine(
        DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    async with _engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield _engine
    async with _engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await _engine.dispose()


@pytest_asyncio.fixture
async def session(engine):
    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with factory() as _session:
        yield _session
        await _session.rollback()   # undo changes between tests

For the full pytest-asyncio integration, scope pitfalls, and the StaticPool shared-connection pattern in detail, see the guide on using aiosqlite for async tests and local development.

Environment-specific engine configuration summary

A practical pattern for multi-environment projects is to centralise engine creation in one factory function that reads from environment variables and returns the correct engine for the current environment:

import os
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.pool import StaticPool, NullPool


def build_engine() -> AsyncEngine:
    """
    Return an appropriately configured AsyncEngine based on DATABASE_URL.
    Handles Postgres (asyncpg/psycopg3), MySQL (aiomysql/asyncmy),
    and SQLite (aiosqlite) from a single factory.
    """
    url = os.environ.get("DATABASE_URL", "sqlite+aiosqlite://")

    if url.startswith("sqlite+aiosqlite://"):
        # In-memory SQLite: StaticPool required; no pool sizing needed
        is_memory = url == "sqlite+aiosqlite://"
        return create_async_engine(
            url,
            connect_args={"check_same_thread": False},
            poolclass=StaticPool if is_memory else StaticPool,
            echo=os.environ.get("SQL_ECHO", "false").lower() == "true",
        )

    if url.startswith("mysql+"):
        # MySQL / MariaDB: explicit recycle to handle wait_timeout
        return create_async_engine(
            url,
            pool_size=int(os.environ.get("DB_POOL_SIZE", "10")),
            max_overflow=int(os.environ.get("DB_MAX_OVERFLOW", "5")),
            pool_pre_ping=True,
            pool_recycle=int(os.environ.get("DB_POOL_RECYCLE", "1800")),
            echo=os.environ.get("SQL_ECHO", "false").lower() == "true",
        )

    # PostgreSQL (asyncpg or psycopg3) — QueuePool default
    return create_async_engine(
        url,
        pool_size=int(os.environ.get("DB_POOL_SIZE", "20")),
        max_overflow=int(os.environ.get("DB_MAX_OVERFLOW", "10")),
        pool_pre_ping=True,
        pool_recycle=int(os.environ.get("DB_POOL_RECYCLE", "1800")),
        echo=os.environ.get("SQL_ECHO", "false").lower() == "true",
    )


# Usage: same import in dev, CI, and production.
# Dev:        DATABASE_URL=sqlite+aiosqlite://
# CI:         DATABASE_URL=postgresql+asyncpg://ci:ci@localhost/ci_db
# Production: DATABASE_URL=postgresql+asyncpg://app:secret@db:5432/prod
engine = build_engine()

This pattern avoids scattering engine configuration across modules and makes the CI→production parity explicit in one function.

Migrating from synchronous drivers to async

The migration path from sync (psycopg2, PyMySQL, sqlite3) to async (asyncpg, aiomysql, aiosqlite) follows the same mechanical steps regardless of backend:

1. Install the async driver package alongside (or instead of) the sync driver.
2. Change create_engine() to create_async_engine() with the updated URL prefix.
3. Replace Session / sessionmaker with AsyncSession / async_sessionmaker.
4. Add await to all session.execute(), session.commit(), session.rollback(), session.close() calls.
5. Wrap session.add() / session.delete() in async with session.begin(): blocks.
6. Replace engine.dispose() (sync) with await engine.dispose() (async).
7. Move Base.metadata.create_all(engine) to async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all).

The most common omission in step 7 is forgetting run_sync — create_all is a synchronous function and must be wrapped:

# Sync (legacy)
Base.metadata.create_all(engine)

# Async (correct)
async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

For dialect-specific quirks that emerge during migrations — particularly asyncpg prepared-statement conflicts with PgBouncer and MySQL charset encoding issues — the guide on dialect-specific gotchas and driver quirks covers the exact error strings and fixes.

Production Pitfalls & Anti-Patterns

• Using QueuePool with an in-memory SQLite database. Each pool checkout creates a new sqlite3 connection, which is a new blank database. Tables created in one connection do not exist in another. Always use poolclass=StaticPool with sqlite+aiosqlite:// so all sessions share one physical connection and thus one schema. Symptom: sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: users on the second session.
• Forgetting pool_recycle with MySQL/MariaDB. Managed MySQL instances (Cloud SQL, RDS, PlanetScale) often set wait_timeout between 600 and 3600 seconds. Without pool_recycle, connections idle in SQLAlchemy's pool past that timeout are silently closed server-side. The next query raises asyncmy.errors.OperationalError: (2013, 'Lost connection to MySQL server during query') or aiomysql.OperationalError: (2006, "MySQL server has gone away"). Fix: pool_recycle=1800 (half the expected timeout) and pool_pre_ping=True.
• Omitting charset=utf8mb4 in MySQL URLs. MySQL's default charset is latin1 on older installations and utf8mb3 on some cloud defaults. Storing emoji, CJK characters, or any 4-byte Unicode codepoints in a latin1 or utf8mb3 column raises Incorrect string value. Always append ?charset=utf8mb4 to your MySQL connection URL.
• Assuming SQLite concurrency for write-heavy tests. SQLite acquires a database-wide write lock per transaction. If multiple pytest workers (via pytest-xdist) share one SQLite file, they will deadlock on concurrent writes. Use an in-memory database per worker (sqlite+aiosqlite://) instead of sharing a file.
• Using asyncpg without pool_pre_ping=True behind a cloud load balancer. AWS ALB, Google Cloud LB, and Azure Front Door all drop idle TCP connections after 350–600 seconds. asyncpg raises asyncpg.exceptions.ConnectionDoesNotExistError on the first query to a stale connection. Set pool_pre_ping=True to issue a lightweight SELECT 1 before handing a connection to the application.
• Running asyncpg in a forked process without NullPool. asyncpg connections are not fork-safe. A parent process that holds open asyncpg connections before os.fork() (common with Gunicorn workers, Celery prefork) leaves children with corrupted connection state. Use poolclass=NullPool and create a fresh engine inside each worker process, or defer engine creation to post-fork hooks.

Frequently Asked Questions

Related

• Async Engines, Dialects, and Connection Pooling — parent reference covering the full async engine and session lifecycle.
• Using aiosqlite for Async Tests and Local Development — StaticPool configuration, pytest-asyncio fixture patterns, and Postgres parity differences.
• Choosing Between asyncpg and psycopg3 Async Drivers — deep dive on the two PostgreSQL driver options: binary protocol, PgBouncer compatibility, and COPY throughput.
• Dialect-Specific Gotchas and Driver Quirks — exact error strings and fixes for PgBouncer prepared-statement conflicts, MySQL charset errors, and SQLite threading issues.
• Configuring Async Engines and Connection Pools — pool_size, max_overflow, pool_recycle, and pre-ping production tuning across all backends.