pyro-mysql 0.2.0

High-performance MySQL driver for Python

pyro-mysql

A high-performance MySQL driver for Python, backed by Rust.

• API Overview
• Usage
• DataType Mapping
• Logging
• PEP-249, sqlalchemy
• Benchmark

Usage

0. Import

# Async
from pyro_mysql.async_ import Conn
from pyro_mysql import AsyncConn

# Sync
from pyro_mysql.sync import Conn
from pyro_mysql import SyncConn

1. Connection

from pyro_mysql.async_ import Conn, Opts

def example():
    conn1 = await Conn.new(f"mysql://{USER}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}")
    conn2 = await Conn.new(
        Opts(f"mysql://{USER}:{PASSWORD}@localhost")
            .tcp_nodelay(false)
    )
    conn3 = await Conn.new(
        Opts()
            .socket("../unix.socket")
            .user("username")
            .db_name("db")
    )

2. Query Execution

SyncConn and AsyncConn provides the following 8 methods (SyncConn doesn't return Awaitable):

# Text Protocols - supports multiple statements concatenated with ';' but accepts no argument
def query(self, query: str, *, as_dict: bool = False) -> Awaitable[list[tuple] | list[dict]]: ...
def query_first(self, query: str, *, as_dict: bool = False) -> Awaitable[tuple | dict | None]: ...
def query_drop(self, query: str) -> Awaitable[None]: ...
def query_batch(self, query: str) -> Awaitable[None]: ...

# Binary Protocols - supports arguments but no multiple statement
def exec(self, query: str, params: Params = None, *, as_dict: bool = False) -> Awaitable[list[tuple] | list[dict]]: ...
def exec_first(self, query: str, params: Params = None, *, as_dict: bool = False) -> Awaitable[tuple | dict | None]: ...
def exec_drop(self, query: str, params: Params = None) -> Awaitable[None]: ...
def exec_batch(self, query: str, params: Iterable[Params] = []) -> Awaitable[None]: ...

# Examples
rows = await conn.exec("SELECT * FROM my_table WHERE a=? AND b=?", (a, b))  # returns list of tuples
rows_as_dicts = await conn.exec("SELECT * FROM my_table WHERE a=? AND b=?", (a, b), as_dict=True)  # returns list of dicts
await conn.exec_batch("SELECT * FROM my_table WHERE a=? AND b=?", [(a1, b1), (a2, b2)])

Awaitable is a coroutine or PyroFuture, which is a Future-like object that tracks a task in the Rust thread. If the returned object is dropped before completion or cancellation, the corresponding task in the Rust thread is cancelled as well.

3. Transaction

# async API
async with conn.start_transaction() as tx:
    await conn.exec('INSERT ..')
    await conn.exec('INSERT ..')
    await tx.commit(conn)

# sync API
with conn.start_transaction() as tx:
    conn.exec('INSERT ..')
    conn.exec('INSERT ..')
    tx.rollback(conn)

4. Choosing a backend

conn = Conn(url)
conn = Conn(url, backend="mysql")
conn = await Conn.new(url, backend="mysql")

• Sync backends: zero (use zero-mysql crate), mysql (use mysql crate)
• Async backends: zero (use zero-mysql crate), mysql (use mysql_async crate)

DataType Mapping

Python -> MySQL

Python Type	MySQL Binary Protocol Encoding
None	NULL
bool	Int64
int	Int64
float	Double(Float64)
str | bytes | bytearray	Bytes
tuple | list | set | frozenset | dict	json-encoded string as Bytes
datetime.datetime	Date(year, month, day, hour, minute, second, microsecond)
datetime.date	Date(year, month, day, 0, 0, 0, 0)
datetime.time	Time(false, 0, hour, minute, second, microsecond)
datetime.timedelta	Time(is_negative, days, hours, minutes, seconds, microseconds)
time.struct_time	Date(year, month, day, hour, minute, second, 0)
decimal.Decimal	Bytes(str(Decimal))
uuid.UUID	Bytes(UUID.hex)

MySQL -> Python

MySQL Column	Python
NULL	None
INT / TINYINT / SMALLINT / MEDIUMINT / BIGINT / YEAR	int
FLOAT	float
DOUBLE	float
DECIMAL / NUMERIC	decimal.Decimal
DATE	datetime.date
TIME	datetime.timedelta
DATETIME	datetime.datetime
TIMESTAMP	datetime.datetime
CHAR / VARCHAR / TEXT / TINYTEXT / MEDIUMTEXT / LONGTEXT	str
BINARY / VARBINARY / BLOB / TINYBLOB / MEDIUMBLOB / LONGBLOB	bytes
JSON	the result of json.loads()
ENUM / SET	str
BIT	bytes
GEOMETRY	bytes (WKB format)

Logging

pyro-mysql sends the Rust logs to the Python logging system, which can be configured with logging.getLogger("pyro_mysql").

# Queries are logged with the DEBUG level
logging.getLogger("pyro_mysql").setLevel(logging.DEBUG)

PEP-249, sqlalchemy

pyro_mysql.dbapi implements PEP-249. This only exists for compatibility with ORM libraries. The primary API set (pyro_mysql.sync, pyro_mysql.async_) is simpler and faster.

pyro_mysql.dbapi
    # classes
    ├─Connection
    ├─Cursor
    # exceptions
    ├─Warning
    ├─Error
    ├─IntegrityError
    ├─..

In sqlalchemy, the following dialects are supported.

• mysql+pyro_mysql:// (sync)
• mariadb+pyro_mysql:// (sync)
• mysql+pyro_mysql_async:// (async)
• mariadb+pyro_mysql_async:// (async)

The supported connection parameters are the docs and capabilities (default 2).

from sqlalchemy import create_engine, text

engine = create_engine("mysql+pyro_mysql://test:1234@localhost/test")
conn = engine.connect()
cursor_result = conn.execute(text("SHOW TABLES"))
for row in cursor_result:
    print(row)

('information_schema',)
('mysql',)
('performance_schema',)
('sys',)
('test',)

To run sqlalchemy tests on pyro_mysql, use this command in the sqlalchemy repo:

pytest -p pyro_mysql.testing.sqlalchemy_pytest_plugin --dburi=mariadb+pyro_mysql://test:1234@localhost/test -v t

sqlalchemy_pytest_plugin is required to skip incompatible tests.

API Overview

• pyro_mysql
• pyro_mysql,sync
• pyro_mysql.async_
• pyro_mysql.dbapi
• pyro_mysql.dbapi_async
• pyro_mysql.error

There is no auto-generated API Reference. *.pyi files are manually synced.

.
└── pyro_mysql/
    ├── (common classes)/
    │   ├── IsolationLevel
    │   ├── CapabilityFlags
    │   └── PyroFuture
    ├── sync/
    │   ├── Conn
    │   ├── Opts
    │   └── Transaction
    ├── async_/
    │   ├── Conn
    │   ├── Opts
    │   └── Transaction
    ├── dbapi/
    │   ├── connect()
    │   ├── Connection
    │   ├── Cursor
    │   └── (exceptions)
    │       ├── Warning
    │       ├── Error
    │       ├── InterfaceError
    │       ├── DatabaseError
    │       ├── DataError
    │       ├── OperationalError
    │       ├── IntegrityError
    │       ├── InternalError
    │       ├── ProgrammingError
    │       └── NotSupportedError
    ├── dbapi_async/
    │   ├── connect()
    │   ├── Connection
    │   ├── Cursor
    │   └── (exceptions)
    │       ├── Warning
    │       ├── Error
    │       ├── InterfaceError
    │       ├── DatabaseError
    │       ├── DataError
    │       ├── OperationalError
    │       ├── IntegrityError
    │       ├── InternalError
    │       ├── ProgrammingError
    │       └── NotSupportedError
    └── (aliases)/
        ├── SyncConn
        ├── SyncTransaction
        ├── AsyncConn
        ├── AsyncTransaction
        └── Opts

Perf Notes

• Prefer MariaDB to MySQL
• Prefer UnixSocket to TCP
• Use BufferPool to reuse allocations between connections
• Use Conn.exec_bulk to group 2~1000 INSERTs or UPDATEs
• Wait for Python 3.14 + mature free-threaded build for faster asyncio performance