tessl/pypi-aiomysql
MySQL driver for asyncio providing async/await support for database operations.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
cursors.mddocs/
Cursors and Data Access
Execute SQL queries and retrieve results using different cursor types optimized for various use cases. aiomysql provides multiple cursor classes to handle different data access patterns efficiently.
Capabilities
Standard Cursor
The default cursor for executing queries and fetching results as tuples.
class Cursor:
@property
def connection(self) -> Connection:
"""Reference to the connection that created this cursor."""
@property
def description(self) -> tuple:
"""
Sequence of 7-item sequences describing result columns.
None if no operation has been performed or no results.
"""
@property
def rowcount(self) -> int:
"""
Number of rows produced or affected by last execute().
-1 if no execute() has been performed.
"""
@property
def rownumber(self) -> int:
"""Current row index (0-based)."""
@property
def arraysize(self) -> int:
"""Default number of rows fetchmany() will return."""
@property
def lastrowid(self) -> int:
"""Row ID of last modified row (for AUTO_INCREMENT columns)."""
@property
def closed(self) -> bool:
"""Whether the cursor is closed."""
async def execute(self, query: str, args = None) -> int:
"""
Execute a SQL query.
Parameters:
- query: SQL query string, may contain %s placeholders
- args: Parameters for query placeholders
Returns:
Number of affected rows
"""
async def executemany(self, query: str, args) -> int:
"""
Execute a SQL query multiple times with different parameters.
Optimized for bulk INSERT operations.
Parameters:
- query: SQL query string with placeholders
- args: Sequence of parameter sequences
Returns:
Number of affected rows
"""
async def callproc(self, procname: str, args = ()) -> tuple:
"""
Call a stored procedure.
Parameters:
- procname: Name of stored procedure
- args: Procedure arguments
Returns:
Modified copy of input arguments
"""
def fetchone(self) -> tuple:
"""
Fetch next row from query results.
Returns:
Row as tuple, or None if no more rows
"""
def fetchmany(self, size: int = None) -> list:
"""
Fetch multiple rows from query results.
Parameters:
- size: Number of rows to fetch (default: arraysize)
Returns:
List of rows as tuples
"""
def fetchall(self) -> list:
"""
Fetch all remaining rows from query results.
Returns:
List of all rows as tuples
"""
def scroll(self, value: int, mode: str = 'relative') -> None:
"""
Scroll cursor position.
Parameters:
- value: Number of rows to move
- mode: 'relative' or 'absolute'
"""
async def nextset(self) -> bool:
"""
Move to next result set (for multi-result queries).
Returns:
True if another result set is available
"""
def mogrify(self, query: str, args = None) -> str:
"""
Format query string with parameters for debugging.
Parameters:
- query: SQL query with placeholders
- args: Parameters for placeholders
Returns:
Formatted query string
"""
async def close(self) -> None:
"""Close the cursor."""Dictionary Cursor
Cursor that returns results as dictionaries with column names as keys.
class DictCursor(Cursor):
"""
Cursor returning results as dictionaries.
Inherits all methods from Cursor with modified return types.
"""
def fetchone(self) -> dict:
"""
Fetch next row as dictionary.
Returns:
Row as dict with column names as keys, or None
"""
def fetchmany(self, size: int = None) -> list:
"""
Fetch multiple rows as dictionaries.
Parameters:
- size: Number of rows to fetch
Returns:
List of dictionaries
"""
def fetchall(self) -> list:
"""
Fetch all rows as dictionaries.
Returns:
List of dictionaries
"""Server-Side Cursor
Unbuffered cursor that fetches results from server on demand, memory-efficient for large result sets.
class SSCursor(Cursor):
"""
Server-side (unbuffered) cursor for large result sets.
Fetch methods are async and retrieve data from server.
"""
async def fetchone(self) -> tuple:
"""
Fetch next row from server.
Returns:
Row as tuple, or None if no more rows
"""
async def fetchmany(self, size: int = None) -> list:
"""
Fetch multiple rows from server.
Parameters:
- size: Number of rows to fetch
Returns:
List of rows as tuples
"""
async def fetchall(self) -> list:
"""
Fetch all remaining rows from server.
Returns:
List of all rows as tuples
"""Server-Side Dictionary Cursor
Combines server-side fetching with dictionary results.
class SSDictCursor(SSCursor):
"""
Server-side cursor returning results as dictionaries.
Combines unbuffered fetching with dictionary format.
"""
async def fetchone(self) -> dict:
"""
Fetch next row from server as dictionary.
Returns:
Row as dict, or None if no more rows
"""
async def fetchmany(self, size: int = None) -> list:
"""
Fetch multiple rows from server as dictionaries.
Parameters:
- size: Number of rows to fetch
Returns:
List of dictionaries
"""
async def fetchall(self) -> list:
"""
Fetch all rows from server as dictionaries.
Returns:
List of dictionaries
"""Usage Examples
Basic Cursor Operations
import asyncio
import aiomysql
async def basic_cursor_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
# Create default cursor (returns tuples)
async with conn.cursor() as cur:
# Execute query
await cur.execute("SELECT id, name, email FROM users WHERE age > %s", (18,))
# Fetch results
print("All users over 18:")
rows = cur.fetchall()
for row in rows:
print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")
conn.close()
asyncio.run(basic_cursor_example())Dictionary Cursor
async def dict_cursor_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
# Use dictionary cursor
async with conn.cursor(aiomysql.DictCursor) as cur:
await cur.execute("SELECT id, name, email FROM users LIMIT 5")
# Results are dictionaries
users = cur.fetchall()
for user in users:
print(f"User: {user['name']} ({user['email']})")
conn.close()Server-Side Cursor for Large Results
async def server_side_cursor_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
# Use server-side cursor for large result set
async with conn.cursor(aiomysql.SSCursor) as cur:
await cur.execute("SELECT * FROM large_table")
# Process results in batches to avoid memory issues
while True:
rows = await cur.fetchmany(1000) # Fetch 1000 rows at a time
if not rows:
break
print(f"Processing batch of {len(rows)} rows")
# Process batch...
conn.close()Server-Side Dictionary Cursor
async def ss_dict_cursor_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
# Best of both: server-side + dictionary results
async with conn.cursor(aiomysql.SSDictCursor) as cur:
await cur.execute("SELECT id, name, created_at FROM logs ORDER BY created_at")
# Stream results as dictionaries
async for row in cur: # Note: this would need to be implemented
print(f"[{row['created_at']}] {row['name']}")
conn.close()Bulk Insert with executemany
async def bulk_insert_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
# Prepare data for bulk insert
user_data = [
('Alice Johnson', 'alice@example.com', 25),
('Bob Smith', 'bob@example.com', 30),
('Carol Davis', 'carol@example.com', 28),
('David Wilson', 'david@example.com', 35)
]
async with conn.cursor() as cur:
# Bulk insert using executemany
await cur.executemany(
"INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
user_data
)
print(f"Inserted {cur.rowcount} users")
# Commit the transaction
await conn.commit()
conn.close()Stored Procedure Call
async def stored_procedure_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
async with conn.cursor() as cur:
# Call stored procedure
result_args = await cur.callproc('get_user_stats', (100,)) # user_id = 100
# Fetch results if procedure returns a result set
stats = cur.fetchall()
print(f"User stats: {stats}")
# Check if there are more result sets
while await cur.nextset():
additional_results = cur.fetchall()
print(f"Additional results: {additional_results}")
conn.close()Query Debugging with mogrify
async def debug_queries_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase'
)
async with conn.cursor() as cur:
query = "SELECT * FROM users WHERE age BETWEEN %s AND %s AND city = %s"
params = (25, 35, 'New York')
# Format query for debugging
formatted_query = cur.mogrify(query, params)
print(f"Executing query: {formatted_query}")
# Execute the actual query
await cur.execute(query, params)
results = cur.fetchall()
print(f"Found {len(results)} users")
conn.close()Install with Tessl CLI
npx tessl i tessl/pypi-aiomysql