asyncmy — The fastest asyncio MySQL/MariaDB driver

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())

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