aiomysql

0.3.2 · active · verified Thu Apr 09

aiomysql is a MySQL driver for asyncio, enabling asynchronous interaction with MySQL databases using Python's `async`/`await` syntax. It provides a familiar DB-API 2.0-like interface adapted for asyncio. The current version is 0.3.2. Releases are infrequent, typically driven by critical bug fixes, `PyMySQL` updates, or community contributions.

Warnings

breaking The `aiomysql.connect` function became an `async def` function in version 0.2.0. Previously, it returned a `Future`.
Fix: Ensure you `await` the `aiomysql.connect()` call directly, e.g., `conn = await aiomysql.connect(...)`. For older versions, you might need `conn = await asyncio.ensure_future(aiomysql.connect(...))` or similar.
gotcha Forgetting to `await` asynchronous operations will lead to `RuntimeWarning: coroutine '...' was never awaited` or unexpected behavior.
Fix: Always prepend `await` to calls like `aiomysql.connect()`, `cursor.execute()`, `cursor.fetchone()`, `pool.acquire()`, and `pool.release()`.
gotcha Not using connection pooling for high-concurrency applications can lead to performance bottlenecks and resource exhaustion.
Fix: For applications with many concurrent database operations, use `await aiomysql.create_pool(...)` to manage a pool of connections. Acquire a connection from the pool with `await pool.acquire()` and release it with `pool.release(conn)` (or use `async with pool.acquire() as conn:`).
gotcha Improperly closing connections and cursors can lead to resource leaks.
Fix: Always use `async with await aiomysql.connect(...) as conn:` and `async with conn.cursor() as cursor:` to ensure connections and cursors are properly closed, even if errors occur.

Install

pip install aiomysql Install aiomysql

Imports

connect

import aiomysql

conn = await aiomysql.connect(...)

DictCursor
```
from aiomysql.cursors import DictCursor
```
While `from aiomysql import DictCursor` often works due to convenience imports, `from aiomysql.cursors import DictCursor` is the explicit and recommended path for cursor types.

create_pool

import aiomysql

pool = await aiomysql.create_pool(...)

Quickstart

This quickstart demonstrates how to connect to a MySQL database, create a table, insert data, and fetch results using `aiomysql`. It uses `async with` statements for robust connection and cursor management, ensuring resources are properly closed. Database credentials are retrieved from environment variables for security.

import asyncio
import os
import aiomysql

async def main():
    # Get credentials from environment variables for security
    db_host = os.environ.get('MYSQL_HOST', '127.0.0.1')
    db_user = os.environ.get('MYSQL_USER', 'root')
    db_password = os.environ.get('MYSQL_PASSWORD', 'password')
    db_name = os.environ.get('MYSQL_DB', 'test_db')

    try:
        # Establish an asynchronous connection
        async with await aiomysql.connect(
            host=db_host,
            user=db_user,
            password=db_password,
            db=db_name,
            autocommit=True # Or manage transactions manually
        ) as conn:
            print(f"Connected to MySQL on {db_host}")

            # Create a cursor object
            async with conn.cursor() as cursor:
                # Execute a query
                await cursor.execute("SELECT VERSION();")
                # Fetch one result
                version = await cursor.fetchone()
                print(f"MySQL Version: {version[0]}")

                # Execute another query (e.g., create a table)
                await cursor.execute(
                    "CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255));"
                )
                print("Table 'users' ensured to exist.")

                # Insert data
                await cursor.execute("INSERT INTO users (name) VALUES (%s);", ("Alice",))
                await cursor.execute("INSERT INTO users (name) VALUES (%s);", ("Bob",))
                print("Inserted Alice and Bob.")

                # Select data
                await cursor.execute("SELECT id, name FROM users;")
                users = await cursor.fetchall()
                print("Users:")
                for user_id, name in users:
                    print(f"  ID: {user_id}, Name: {name}")

            # The connection is automatically closed when exiting the 'async with' block

    except Exception as e:
        print(f"An error occurred: {e}")

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

view raw JSON →