asyncmy2

0.2.20 verified Mon Apr 27 auth: no python

A fast asyncio MySQL driver with native Python coroutines and automatic connection pooling. Version 0.2.20, requires Python >=3.10, actively maintained.

pip install asyncmy2

Common errors

error ModuleNotFoundError: No module named 'asyncmy2' ↓

cause Package not installed or installed as 'asyncmy2' under a different name.

fix

Run: pip install asyncmy2

error asyncmy2.exceptions.OperationalError: (1045, "Access denied for user '...'@'...' (using password: YES)") ↓

cause Incorrect MySQL credentials.

fix

Verify your MySQL user, password, and host. Ensure the user has permissions.

error RuntimeError: Event loop is closed ↓

cause Trying to use asyncmy2 after the event loop is closed or when using asyncio.run() incorrectly with nested loops.

fix

Use asyncio.run() only once. Do not call asyncmy2 functions after the loop has closed. Create connections inside the main async function.

Warnings

breaking asyncmy2 is not a drop-in replacement for asyncmy. The import paths and API differ. Do not mix asyncmy2 with asyncmy in the same project. ↓

fix Use consistent imports from asyncmy2 only.

gotcha Connections must be explicitly closed with `await conn.ensure_closed()`. Not doing so can leak connections, especially if using connection pools. ↓

fix Always call `await conn.ensure_closed()` after use, or use context managers if available.

gotcha The `Pool` class is not re-entrant. You cannot create a pool inside an already running event loop without proper shutdown. Use `await pool.close()` before the loop ends. ↓

fix Ensure `await pool.close()` is called during cleanup.

Imports

connect
```
from asyncmy2 import connect
```
asyncmy uses 'asyncmy.connect', but asyncmy2 moved to top-level
Pool
```
from asyncmy2.pool import Pool
```
Pool is in asyncmy2.pool, not asyncmy.pool

Quickstart

Connect to MySQL and run a simple query.

import asyncio
from asyncmy2 import connect

async def main():
    conn = await connect(host='127.0.0.1', port=3306, user='user', password='pass', db='test')
    async with conn.cursor() as cur:
        await cur.execute("SELECT 1")
        result = await cur.fetchone()
        print(result)
    await conn.ensure_closed()

asyncio.run(main())