asyncmy is the fastest asyncio MySQL/MariaDB driver for Python. It keeps the familiar aiomysql API while rewriting the entire protocol core in Cython — down to pointer-level packet parsing. In our benchmarks it outperforms every driver tested, including the C-based synchronous mysqlclient.

Features

• 🚀 Fastest in every benchmark — reads large result sets 2.1x faster than mysqlclient and 5x faster than aiomysql/pymysql (details)
• 🔌 Drop-in aiomysql replacement — same API, same cursors (DictCursor, SSCursor), same pool semantics
• 🧬 Server-side prepared statements (binary protocol) via conn.prepare() — no client-side escaping, no text parsing; large scans another ~35% faster than the text protocol
• ⚡ C-speed protocol core — rows are parsed in bulk from the receive buffer in a single C loop, values decode straight from wire bytes via the CPython C-API
• 🏊 Built-in connection pool — asyncmy.create_pool(), no extra dependency, 2x aiomysql's pooled throughput
• 📡 MySQL replication protocol over asyncio (BinLogStream)
• ✅ CI-tested on MySQL and MariaDB (workflow)

Benchmark

asyncmy ranks #1 in all four scenarios against mysqlclient, pymysql, and aiomysql (warmup + best-of-3, see methodology):

The protocol core is engineered for zero waste on the hot path:

• Bulk packet parsing: one socket read serves hundreds of row packets, parsed in a single C loop with no event-loop round-trips
• Pointer-based protocol reads: integers and length-encoded values are read directly from raw memory, no struct calls
• Direct row decoding: cell values decode straight from the receive buffer via the CPython C-API (PyUnicode_DecodeUTF8, PyTuple_New), skipping intermediate objects
• Zero-decode numeric/temporal columns: int/float/datetime values parse directly from bytes, and dates are built with the C datetime API
• Escape fast path: strings without special characters are returned as-is, no translation pass

📊 View detailed benchmarks →

Install

Requirements: Python ≥ 3.9

pip install asyncmy

Windows

asyncmy uses Cython extensions; on Windows you need Microsoft C++ Build Tools to build them.

1. Download Microsoft C++ Build Tools.

2. Open CMD as Administrator (recommended) and cd to the folder where the installer was downloaded.

3. Rename the installer (e.g. vs_buildtools__XXXXXXXXX.XXXXXXXXXX.exe) to vs_buildtools.exe for convenience.

4. Run (ensure ~5–6GB free disk space):

   vs_buildtools.exe --norestart --passive --downloadThenInstall --includeRecommended --add Microsoft.VisualStudio.Workload.NativeDesktop --add Microsoft.VisualStudio.Workload.VCTools --add Microsoft.VisualStudio.Workload.MSBuildTools

5. Wait for installation to complete, then restart your computer.

6. Install asyncmy:

   pip install asyncmy

You can uninstall the Build Tools afterward if desired.

Usage

connect

Use asyncmy.connect() for a single connection. For many concurrent connections, use a connection pool.

import asyncio
import os
from asyncmy import connect
from asyncmy.cursors import DictCursor
async def main():
    conn = await connect(
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD", ""),
    )
    async with conn.cursor(cursor=DictCursor) as cursor:
        await cursor.execute("CREATE DATABASE IF NOT EXISTS test")
        await cursor.execute("""
            CREATE TABLE IF NOT EXISTS test.`asyncmy` (
                `id`       int PRIMARY KEY AUTO_INCREMENT,
                `decimal`  decimal(10, 2),
                `date`     date,
                `datetime` datetime,
                `float`    float,
                `string`   varchar(200),
                `tinyint`  tinyint
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
        """.strip())
    await conn.ensure_closed()
if __name__ == "__main__":
    asyncio.run(main())

Prepared statements (binary protocol)

For repeated queries, server-side prepared statements skip client-side escaping entirely and read results in MySQL's binary protocol — numeric and temporal columns decode natively with no text parsing. Placeholders use native ? syntax.

stmt = await conn.prepare("SELECT id, name FROM users WHERE id = ?")
result = await stmt.execute((42,))
print(result.rows)           # tuple of row tuples
print(result.affected_rows)  # for INSERT/UPDATE/DELETE
await stmt.close()
# or as a context manager
async with await conn.prepare("SELECT ? + ?") as stmt:
    result = await stmt.execute((1, 2))

Transparent mode: pass stmt_cache_size=N to connect()/create_pool() and regular cursor.execute("... %s ...", args) calls automatically run as cached server-side prepared statements — no code changes needed (ORMs benefit too). Queries the server can't prepare fall back to the text protocol silently.

pool = await asyncmy.create_pool(stmt_cache_size=128, ...)

Note: with the binary protocol, FLOAT columns return the exact stored value rather than the text protocol's decimal-rounded rendering, which is why this is opt-in.

Pool

For multiple connections, use a connection pool. Pass the same kwargs as connect() (e.g. host, user, password).

import asyncio
import asyncmy
async def main():
    pool = await asyncmy.create_pool(host="localhost", user="root", password="")
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute("SELECT 1")
            ret = await cursor.fetchone()
            assert ret == (1,)
    pool.close()
    await pool.wait_closed()
if __name__ == "__main__":
    asyncio.run(main())

Type checking

asyncmy ships py.typed and stubs for its compiled modules, so mypy and Pylance resolve the API without extra configuration:

conn = await asyncmy.connect(host="localhost", user="root")   # -> Connection
async with conn.cursor() as cur:
    rows = await cur.fetchall()                               # -> list[Any]

Contributors: make stubs regenerates the stubs after changing a .pyx signature. make check runs stubtest, which compares every stub against the compiled module and fails on drift.

Rotating credentials

Some credentials expire while a pooled connection outlives them — AWS RDS IAM auth tokens last 15 minutes, for instance. Pass password_creator instead of password and it is consulted before every connection attempt, including the ones the pool makes on its own when it recycles or reconnects:

import boto3
client = boto3.client("rds")
def rds_auth_token():
    return client.generate_db_auth_token(
        DBHostname="mydb.cluster.amazonaws.com",
        Port=3306,
        DBUsername="dbuser",
        Region="us-east-1",
    )
pool = await asyncmy.create_pool(
    host="mydb.cluster.amazonaws.com",
    user="dbuser",
    password_creator=rds_auth_token,
)

The callable may be a plain function or return an awaitable, and must return str or bytes. If both password and password_creator are given, the creator wins.

Statement logging

echo=True logs every statement and its duration to the asyncmy logger at INFO level. Nothing appears until logging is configured — logging.basicConfig(level=logging.INFO) at minimum, since the root logger defaults to WARNING.

For anything beyond that, pass query_callback. It is called as callback(cursor, query, elapsed_ms) after every successful statement, with the duration as a float in milliseconds:

import logging
logger = logging.getLogger("myapp.sql")
def log_slow_queries(cursor, query, elapsed_ms):
    if elapsed_ms > 100:
        logger.warning("[%sms] %s", elapsed_ms, query)
conn = await asyncmy.connect(host="localhost", user="root", query_callback=log_slow_queries)

executemany and callproc report once for the whole call rather than once per row. The callback is independent of echo: set both and you get the log line and the callback.

Replication

asyncmy supports the MySQL replication protocol (like python-mysql-replication) over asyncio.

import asyncio
from asyncmy import connect
from asyncmy.replication import BinLogStream
async def main():
    conn = await connect()
    ctl_conn = await connect()
    stream = BinLogStream(
        conn,
        ctl_conn,
        server_id=1,
        master_log_file="binlog.000172",
        master_log_position=2235312,
        resume_stream=True,
        blocking=True,
    )
    async for event in stream:
        print(event)
    await conn.ensure_closed()
    await ctl_conn.ensure_closed()
if __name__ == "__main__":
    asyncio.run(main())

Acknowledgments

asyncmy builds on these projects:

• PyMySQL — pure Python MySQL client
• aiomysql — asyncio MySQL driver
• python-mysql-replication — MySQL replication protocol (pure Python, on top of PyMySQL)

License

Apache-2.0