# asyncmy

**Repository Path**: irvinren/asyncmy

## Basic Information

- **Project Name**: asyncmy
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: dev
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-02-01
- **Last Updated**: 2026-09-11

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# asyncmy — The fastest asyncio MySQL/MariaDB driver

[![PyPI](https://img.shields.io/pypi/v/asyncmy.svg)](https://pypi.org/pypi/asyncmy)
[![License](https://img.shields.io/github/license/long2ice/asyncmy)](https://github.com/long2ice/asyncmy)
[![CI](https://github.com/long2ice/asyncmy/actions/workflows/ci.yml/badge.svg)](https://github.com/long2ice/asyncmy/actions/workflows/ci.yml)
[![Release](https://github.com/long2ice/asyncmy/actions/workflows/pypi.yml/badge.svg)](https://github.com/long2ice/asyncmy/actions/workflows/pypi.yml)

`asyncmy` is the fastest asyncio MySQL/MariaDB driver for Python. It keeps the familiar [aiomysql](https://github.com/aio-libs/aiomysql) API while rewriting the entire protocol core in [Cython](https://cython.org/) — down to pointer-level packet parsing. In [our benchmarks](./benchmark/README.md) 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](./benchmark/README.md))
- 🔌 **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](https://github.com/long2ice/asyncmy/blob/dev/asyncmy/replication/binlogstream.py))
- ✅ **CI-tested on MySQL and MariaDB** ([workflow](https://github.com/long2ice/asyncmy/blob/dev/.github/workflows/ci.yml))

## Benchmark

asyncmy ranks **#1 in all four scenarios** against `mysqlclient`, `pymysql`, and `aiomysql` (warmup + best-of-3, see [methodology](./benchmark/README.md#methodology)):

| Test | asyncmy Rank | Performance |
| ---- | ------------ | ----------- |
| **Large Result Set** (33k rows, all types) | 🏆 **#1/4** | 0.030s — 2.2x faster than mysqlclient, 5.3x faster than aiomysql |
| **Connection Pool** (2k queries) | 🏆 **#1/2** | ~17,000 qps — 2x aiomysql's throughput |
| **Concurrent Queries** (50 connections) | 🏆 **#1/2** | ~8,000 qps — 1.6x faster than aiomysql |
| **Batch Insert** (10k rows) | 🏆 **#1/4** | ~107,000 rows/sec — fastest of all four drivers |

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 →](./benchmark/README.md)**

## Install

**Requirements:** Python ≥ 3.9

```bash
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](https://visualstudio.microsoft.com/visual-cpp-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):

   ```bash
   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:

   ```bash
   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](#pool).

```py
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.

```py
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.

```py
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`).

```py
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:

```py
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:

```py
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:

```py
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](https://github.com/noplay/python-mysql-replication)) over asyncio.

```py
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](https://github.com/PyMySQL/PyMySQL) — pure Python MySQL client
- [aiomysql](https://github.com/aio-libs/aiomysql) — asyncio MySQL driver
- [python-mysql-replication](https://github.com/noplay/python-mysql-replication) — MySQL replication protocol (pure Python, on top of PyMySQL)

## License

[Apache-2.0](./LICENSE)