aiomysql

raw JSON →
0.3.2 verified Thu Apr 23 auth: no python install: verified

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.

pip install aiomysql

Common errors

error pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'") ↓
cause This error occurs when the MySQL server is unreachable at the specified host and port.
fix
Ensure the MySQL server is running and accessible at '127.0.0.1' on port 3306, and verify that the connection parameters are correct.
error RuntimeError: Event loop is closed ↓
cause This error occurs when an operation is attempted on a closed asyncio event loop.
fix
Ensure that the event loop is open and running before performing asynchronous operations.
error pymysql.err.InternalError: Packet sequence number wrong - got 1 expected 2 ↓
cause This error indicates a mismatch in the packet sequence numbers during communication with the MySQL server.
fix
Check for network issues or bugs in the client-server communication that may cause packet loss or duplication.
error aiomysql raise InvalidRequestError: Cannot release a connection with not finished transaction ↓
cause This error occurs when attempting to release a connection that has an unfinished transaction.
fix
Ensure that all transactions are properly committed or rolled back before releasing the connection.
error aiomysql does not work with ssl, but pymysql does work ↓
cause This issue arises when aiomysql fails to establish an SSL connection, whereas pymysql succeeds.
fix
Verify the SSL configuration and compatibility between aiomysql and the MySQL server.

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.
gotcha Failed to connect to the MySQL server. This can be caused by the server not running, incorrect host/port, firewall issues, or network configuration. ↓
fix Ensure your MySQL server is running and accessible from where your application is running. Verify the host, port, username, and password in your `aiomysql.connect()` parameters. Check firewall rules and network connectivity between your application and the database server.
gotcha The application failed to connect to the MySQL server. This error typically indicates that the database server is not running, is not accessible from the specified host and port, or network/firewall rules are preventing the connection. ↓
fix Ensure the MySQL server is running, configured to accept connections from the application's host and port, and that no network or firewall rules are blocking the connection.

Install compatibility verified last tested: 2026-04-23

runtime status import time mem disk
3.10-alpine 0.13s 5.1MB 18.7M
3.10-slim 0.09s 5.1MB 19M
3.11-alpine 0.23s 6.3MB 20.7M
3.11-slim 0.18s 6.3MB 21M
3.12-alpine 0.46s 9.2MB 12.6M
3.12-slim 0.42s 9.2MB 13M
3.13-alpine 0.46s 9.7MB 12.2M
3.13-slim 0.42s 9.7MB 13M
3.9-alpine 0.13s 5.0MB 18.2M
3.9-slim 0.12s 5.0MB 19M

Imports

Quickstart last tested: 2026-04-24

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())
last verified Thu Apr 23
next check Wed Jul 08