Choosing between aiomysql and asyncmy for async MySQL

Use mysql+asyncmy:// when you can install its compiled wheels and read large result sets, and mysql+aiomysql:// when you need a pure-Python install — both give SQLAlchemy the same API, and either way set pool_recycle, pool_pre_ping=True and charset=utf8mb4. This guide belongs to selecting async drivers for SQLite, MySQL and Postgres.

Quick Answer

The driver decision is smaller than the configuration around it. Most async MySQL incidents come from idle connections and character sets, not from which driver parsed the rows.

Before — a minimal URL that fails in production:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("mysql+aiomysql://shop:secret@db/shop")
# Hours later, after a quiet period:
# sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError)
# (2013, 'Lost connection to MySQL server during query')
# And on the first product name with an emoji:
# (1366, "Incorrect string value: '\xF0\x9F\x8E\x81' for column 'name' at row 1")

After — either driver, configured for production:

import ssl

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(
    "mysql+asyncmy://shop:secret@db:3306/shop?charset=utf8mb4",
    pool_size=10,
    max_overflow=5,
    pool_recycle=1800,        # seconds; below every idle timeout on the path
    pool_pre_ping=True,       # test a connection before handing it out
    connect_args={
        "ssl": ssl.create_default_context(),
        "init_command": "SET time_zone = '+00:00'",
    },
)
Session = async_sessionmaker(engine, expire_on_commit=False)

Swapping asyncmy for aiomysql in that URL is the entire difference between the two configurations. The init_command pins the session time zone so NOW() and TIMESTAMP conversions do not depend on the server's default.

Execution Context & Async Workflow Integration

Both drivers descend from PyMySQL, the pure-Python MySQL client. aiomysql wraps PyMySQL's protocol code with asyncio I/O and is maintained under the aio-libs organisation; its strength is that it installs anywhere Python does. asyncmy reimplements the same protocol in Cython, which makes decoding rows — the CPU-bound part of reading a large result — substantially faster, at the cost of needing a compiled wheel for your platform or a compiler at install time.

SQLAlchemy drives both through the same mysql dialect with a thin async adapter, so everything above the driver is identical: the AsyncSession API, the SQL generated for each construct, the exception hierarchy you catch, and the pool. Exception origins differ — exc.orig wraps a pymysql.err class under aiomysql and an asyncmy.errors class under asyncmy — but both carry the MySQL error number as the first argument, which is the portable thing to inspect:

from sqlalchemy.exc import DBAPIError

DUPLICATE_ENTRY = 1062
LOST_CONNECTION = {2006, 2013}


def mysql_errno(exc: DBAPIError) -> int | None:
    args = getattr(exc.orig, "args", ())
    return args[0] if args and isinstance(args[0], int) else None

The operational behaviour that matters most is how MySQL treats idle connections. The server closes any connection idle for longer than wait_timeout — eight hours by default, often much less on managed services — and cloud load balancers and proxies in front of MySQL frequently close idle TCP connections after a few minutes. SQLAlchemy's pool cannot see that happen. The next checkout receives a dead connection, and the first query fails with error 2013, Lost connection to MySQL server during query, or 2006, MySQL server has gone away.

Two pool settings prevent it. pool_recycle discards connections older than the given number of seconds at checkout, so set it below the shortest idle timeout between the application and the server. pool_pre_ping runs a cheap ping at checkout and replaces the connection transparently if it fails, which also covers failovers and restarts that recycling cannot predict. The same reasoning for PostgreSQL is in configuring pool_pre_ping to handle stale connections.

Resolving Warnings, Errors & Common Mistakes

The character-set row catches applications long after launch, when the first user types an emoji. MySQL's utf8 is a legacy three-byte encoding that cannot store characters outside the Basic Multilingual Plane; utf8mb4 is real UTF-8. Both the connection and the column must use utf8mb4, which means the URL parameter and the table definitions — mysql_charset="utf8mb4" in __table_args__, or a server default of utf8mb4 — have to agree.

The shutdown error is noisier with the MySQL drivers than with asyncpg, because their connection finalisers try to use the event loop. It is harmless at exit and entirely avoided by disposing the engine inside the application's lifespan, as covered in fixing garbage collector non-checked-in connection warnings.

Advanced: Streaming, Bulk Inserts and MySQL-Specific Behaviour

Large reads are where the drivers differ most, and where MySQL's protocol shapes the code. By default a MySQL client receives an entire result set into memory before the first row is returned. Streaming requires a server-side cursor, which SQLAlchemy requests with stream_results, and AsyncSession.stream() sets it for you:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Order


async def export_orders(session: AsyncSession, writer) -> int:
    count = 0
    result = await session.stream_scalars(
        select(Order).order_by(Order.id).execution_options(yield_per=2_000)
    )
    async with result:
        async for order in result:
            await writer.write_row(order)
            count += 1
    return count

A MySQL server-side cursor holds the connection busy until the result is fully read or closed — no other statement can run on that connection meanwhile — so keep the loop body free of other queries on the same session, and close the result promptly. With large exports, asyncmy's compiled row decoding shows up clearly in wall-clock time; with aiomysql, CPU spent decoding rows is the usual bottleneck.

Bulk inserts meet MySQL's lack of RETURNING. SQLAlchemy 2.0's fast "insertmanyvalues" path batches multi-row inserts and, on databases with RETURNING, recovers generated keys in the same statements. MySQL has no RETURNING, so when the ORM needs every new primary key it must fall back to inserting rows individually and reading lastrowid. Two ways around it: use Core insert() with a list of dictionaries when you do not need keys back, or generate keys in the application — UUIDs, or IDs from a sequence service — so the ORM never has to ask. MariaDB 10.5 and later supports INSERT ... RETURNING, and SQLAlchemy uses it with the mariadb dialect (mariadb+asyncmy://).

Three more server behaviours deserve attention when coming from PostgreSQL. The default isolation level is REPEATABLE READ, so a long transaction sees a snapshot from its first read; set isolation_level="READ COMMITTED" on the engine if code assumes PostgreSQL's default, as discussed in setting transaction isolation level per session. DDL statements commit implicitly, so an Alembic migration that fails halfway leaves the earlier statements applied. And upserts use INSERT ... ON DUPLICATE KEY UPDATE, available as sqlalchemy.dialects.mysql.insert(...).on_duplicate_key_update(...).

Testing and Switching Between the Two Drivers

Because SQLAlchemy hides the driver, the practical way to choose is to run your own workload on both and compare — and to keep the option of switching open by never importing either driver directly.

Keep driver names in configuration only. Error handling that inspects MySQL error numbers through exc.orig.args, as above, works for both; code that imports pymysql.err or asyncmy.errors pins the choice. A single environment variable for the URL scheme then makes a comparison run trivial:

import os

from sqlalchemy.ext.asyncio import create_async_engine

DRIVER = os.environ.get("MYSQL_ASYNC_DRIVER", "asyncmy")   # or "aiomysql"

engine = create_async_engine(
    f"mysql+{DRIVER}://shop:secret@db:3306/shop?charset=utf8mb4",
    pool_recycle=1800,
    pool_pre_ping=True,
)

Run the integration suite under both values in CI for a while. Differences are rare, and the ones that appear tend to be in edge cases worth knowing about anyway — type conversions for DECIMAL and BIT columns, how JSON columns are returned, and error messages that tests match on as strings.

For performance, measure the queries that matter rather than synthetic benchmarks. A short script that runs your three heaviest reads and one bulk write a few hundred times under each driver, against a MySQL instance on the same network as production, tells you more than any published comparison. Expect the difference to concentrate in large result sets and to disappear for small lookups, where network latency dominates.

Local development is the one place neither driver is the right answer for every test. Tests that exercise MySQL-specific behaviour — ON DUPLICATE KEY UPDATE, REPEATABLE READ snapshots, implicit DDL commits — need a real MySQL, typically in a container. Tests that exercise only portable SQLAlchemy constructs can often run faster on SQLite, with the caveats in using aiosqlite for async tests and local development. Mixing the two is common; just be explicit about which tests need which.

Frequently Asked Questions

Related

• Selecting Async Drivers for SQLite, MySQL and Postgres — The parent guide: driver options across databases.
• Using aiosqlite for async tests and local development — When SQLite is enough for tests, and when it is not.
• Configuring pool_pre_ping to handle stale connections — The checkout test that covers idle-timeout disconnects.
• Choosing between asyncpg and psycopg async drivers — The equivalent decision for PostgreSQL.