asyncmy — Fast asyncio MySQL/MariaDB driver

asyncmy is a fast asyncio MySQL/MariaDB driver. It reuses most of PyMySQL and aiomysql while rewriting the core protocol in Cython for better performance.

Features

• API compatible with aiomysql
• Faster via Cython-compiled core
• MySQL replication protocol with asyncio (BinLogStream)
• CI-tested on MySQL and MariaDB (workflow)

Benchmark

asyncmy demonstrates excellent performance across realistic workloads:

Test	asyncmy Rank	Performance
Connection Pool (2k queries)	🏆 #1/2	~10,500 qps (consistently 22-28% faster than aiomysql)
Large Result Set (50k rows)	#2/4	~0.090s (2x faster than aiomysql, close to mysqlclient)
Concurrent Queries (50 queries)	#1-2/2	Comparable to aiomysql
Batch Insert (10k rows)	Variable	Results vary by run

Recent optimizations (v0.2.12) delivered significant performance improvements:

• Buffer Management: Zero-copy fast path for single-packet reads
• DateTime Parsing: Fast string slicing replacing regex
• Row Parsing: Pre-allocated lists and C-level indexing in hot path
• Protocol Parsing: Inlined length-coded string reads with fast path for common cases

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

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