asyncmy - A fast asyncio MySQL Driver

Warnings

• gotcha On Windows, `asyncmy` uses Cython extensions which require Microsoft C++ Build Tools to be installed for successful installation via pip. Without these tools, installation may fail or result in a non-optimized pure-Python fallback.
  Fix: Download and install Microsoft C++ Build Tools from the Visual Studio website before running `pip install asyncmy`.
• gotcha Always ensure connections and cursors are properly closed to prevent resource leaks. While `async with` statements handle this for contexts, direct `connect()` calls require an explicit `await conn.close()`.
  Fix: Use `async with conn.cursor(...)` and `async with asyncmy.create_pool(...)` where possible. For direct `connect()`, always pair with `await conn.close()` in a `finally` block.
• gotcha Common `asyncio` pitfalls like 'fire and forget' tasks (not awaiting coroutines) or blocking the event loop with synchronous operations can lead to unexpected behavior, lost exceptions, or reduced performance. `asyncmy` operations are all awaitable and should be used within an `asyncio` event loop.
  Fix: Ensure all `async` functions are `await`ed. Use `asyncio.create_task()` for background tasks and `asyncio.gather()` for concurrent execution. Avoid blocking I/O or CPU-bound synchronous calls directly in the event loop.
• breaking The `loop` argument for `asyncmy.connect()` and `asyncmy.create_pool()` was removed in version 0.2.1, as passing an explicit event loop is generally no longer necessary or recommended in modern `asyncio`.
  Fix: Remove the `loop=asyncio.get_event_loop()` argument from `connect()` and `create_pool()` calls. `asyncmy` will automatically use the current running event loop.

Install

• pip install asyncmy Standard Install
• pip install asyncmy # On Windows, requires Microsoft C++ Build Tools Windows Install

Imports

• connect

  from asyncmy import connect

• create_pool

  from asyncmy import create_pool

  For managing multiple database connections efficiently.
• DictCursor

  from asyncmy.cursors import DictCursor

  A cursor that returns rows as dictionaries.
• BinLogStream

  from asyncmy.replication import BinLogStream

  For working with MySQL's replication protocol.

Quickstart

This example demonstrates connecting to a MySQL database, executing DDL and DML operations, and fetching results using a DictCursor. It uses environment variables for connection parameters and ensures proper resource cleanup using `async with` and a `finally` block.

import asyncio
import os
from asyncmy import connect
from asyncmy.cursors import DictCursor

async def main():
    conn = await connect(
        host=os.environ.get('MYSQL_HOST', '127.0.0.1'),
        user=os.environ.get('MYSQL_USER', 'root'),
        password=os.environ.get('MYSQL_PASSWORD', ''),
        db=os.environ.get('MYSQL_DB', 'test_db'),
        port=int(os.environ.get('MYSQL_PORT', 3306))
    )
    try:
        async with conn.cursor(cursor=DictCursor) as cursor:
            await cursor.execute("CREATE DATABASE IF NOT EXISTS test_db")
            await cursor.execute("USE test_db")
            await cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id INT PRIMARY KEY AUTO_INCREMENT,
                    name VARCHAR(255),
                    email VARCHAR(255)
                )
                """
            )
            await cursor.execute(
                "INSERT INTO users (name, email) VALUES (%s, %s)",
                ("Alice", "alice@example.com")
            )
            await cursor.execute("SELECT id, name, email FROM users WHERE name = %s", ("Alice",))
            result = await cursor.fetchone()
            print(f"Fetched user: {result}")
    finally:
        await conn.close()

if __name__ == '__main__':
    asyncio.run(main())