or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

connections.mdcursors.mdindex.mdpooling.mdsqlalchemy.md
tile.json

tessl/pypi-aiomysql

MySQL driver for asyncio providing async/await support for database operations.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/aiomysql@0.2.x

To install, run

npx @tessl/cli install tessl/pypi-aiomysql@0.2.0

index.mddocs/

aiomysql

A pure-Python MySQL client library providing async/await support for asyncio applications. Built on top of PyMySQL, aiomysql enables non-blocking database operations while maintaining compatibility with the MySQL protocol and offering connection pooling capabilities.

Package Information

  • Package Name: aiomysql
  • Language: Python
  • Installation: pip install aiomysql
  • Dependencies: PyMySQL>=1.0, optional SQLAlchemy support

Core Imports

import aiomysql

Individual components:

from aiomysql import connect, create_pool, Connection, Pool
from aiomysql import Cursor, DictCursor, SSCursor, SSDictCursor

Basic Usage

import asyncio
import aiomysql

async def main():
    # Connect to MySQL database
    conn = await aiomysql.connect(
        host='localhost',
        port=3306,
        user='root',
        password='password',
        db='test'
    )
    
    # Create cursor and execute query
    async with conn.cursor() as cur:
        await cur.execute("SELECT 42")
        result = await cur.fetchone()
        print(result)  # (42,)
    
    # Close connection
    conn.close()

# Run the example
asyncio.run(main())

Architecture

aiomysql provides two main integration approaches:

  • Direct MySQL Integration: Low-level async MySQL connectivity with cursors, connections, and pools
  • SQLAlchemy Integration: High-level ORM-style operations through the aiomysql.sa subpackage

The library maintains full compatibility with MySQL and MariaDB servers while providing async/await support for all database operations including transactions, connection pooling, and different cursor types for various data access patterns.

Capabilities

Database Connections

Create and manage individual MySQL database connections with support for SSL, authentication plugins, and connection configuration.

def connect(
    host: str = "localhost",
    user: str = None,
    password: str = "",
    db: str = None,
    port: int = 3306,
    unix_socket: str = None,
    charset: str = '',
    sql_mode: str = None,
    read_default_file: str = None,
    conv: dict = None,
    use_unicode: bool = None,
    client_flag: int = 0,
    cursorclass: type = Cursor,
    init_command: str = None,
    connect_timeout: int = None,
    read_default_group: str = None,
    autocommit: bool = False,
    echo: bool = False,
    local_infile: bool = False,
    loop = None,
    ssl: dict = None,
    auth_plugin: str = '',
    program_name: str = '',
    server_public_key: str = None
) -> _ConnectionContextManager: ...

Database Connections

Connection Pooling

Manage multiple database connections efficiently with connection pooling, supporting minimum and maximum pool sizes, connection recycling, and pool lifecycle management.

def create_pool(
    minsize: int = 1,
    maxsize: int = 10,
    echo: bool = False,
    pool_recycle: int = -1,
    loop = None,
    **kwargs
) -> _PoolContextManager: ...

Connection Pooling

Cursors and Data Access

Execute SQL queries and retrieve results using different cursor types optimized for various use cases, from simple queries to large result sets and dictionary-style data access.

class Cursor:
    async def execute(self, query: str, args = None) -> int: ...
    async def executemany(self, query: str, args) -> int: ...
    def fetchone(self) -> tuple: ...
    def fetchmany(self, size: int = None) -> list: ...
    def fetchall(self) -> list: ...

class DictCursor(Cursor):
    def fetchone(self) -> dict: ...
    def fetchmany(self, size: int = None) -> list: ...
    def fetchall(self) -> list: ...

class SSCursor(Cursor):
    async def fetchone(self) -> tuple: ...
    async def fetchmany(self, size: int = None) -> list: ...
    async def fetchall(self) -> list: ...

class SSDictCursor(SSCursor):
    async def fetchone(self) -> dict: ...
    async def fetchmany(self, size: int = None) -> list: ...
    async def fetchall(self) -> list: ...

Cursors and Data Access

SQLAlchemy Integration

High-level database operations using SQLAlchemy's expression language and ORM capabilities with async/await support through the aiomysql.sa subpackage.

async def create_engine(
    minsize: int = 1,
    maxsize: int = 10,
    loop = None,
    dialect = None,
    pool_recycle: int = -1,
    compiled_cache = None,
    **kwargs
) -> Engine: ...

SQLAlchemy Integration

Exception Types

# Base exceptions
class Error(Exception): ...
class Warning(Exception): ...

# Interface and database errors  
class InterfaceError(Error): ...
class DatabaseError(Error): ...
class DataError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class IntegrityError(DatabaseError): ...
class InternalError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...
class NotSupportedError(DatabaseError): ...

# MySQL-specific error
class MySQLError(Exception): ...

Context Manager Types

class _ConnectionContextManager:
    """Context manager for database connections."""
    async def __aenter__(self) -> Connection: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

class _PoolContextManager:
    """Context manager for connection pools."""
    async def __aenter__(self) -> Pool: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...

Utility Functions

def escape_string(s: str) -> str: ...
def escape_dict(d: dict, charset: str) -> dict: ...
def escape_sequence(t: tuple, charset: str) -> tuple: ...