repository·main·Indexed 23 days ago
https://github.com/aio-libs/aiomysqlAn asynchronous MySQL driver for Python's asyncio framework, designed as a drop-in async replacement for PyMySQL with an API similar to aiopg. It provides functionality for establishing direct connections via connect(), managing connection pools with create_pool(), and utilizing various cursor types including Cursor, DictCursor, SSCursor (server-side), and SSDictCursor. The library includes support for transaction management, bulk inserts via executemany(), and automatic JSON parsing through DeserializationCursor.
asyncio framework. It is built by extending PyMySQL, replacing underlying I/O calls with asynchronous versions. It aims to provide an API and developer experience similar to the aiopg library.The aiomysql.Connection object represents a socket with a MySQL server. Its interface is nearly identical to pymysql.connection, but all methods are coroutines and must be awaited.
Key capabilities include:
cursor().begin(), commit(), and rollback().autocommit(value) and get_autocommit().select_db(db).close() or ensure_closed().Transactions in aiomysql.sa are managed via Transaction objects obtained from an SAConnection. You control transaction boundaries using .commit() and .rollback().
Use await conn.begin() to start a transaction.
async with engine.acquire() as conn:
trans = await conn.begin()
try:
await conn.execute("insert into x (a, b) values (1, 2)")
except Exception:
await trans.rollback()
else:
await trans.commit()Use await conn.begin_nested() to create a NestedTransaction. This represents a MySQL SAVEPOINT. The interface is identical to a standard Transaction.
Use await conn.begin_twophase() to create a TwoPhaseTransaction. This is used for distributed transactions. In addition to commit() and rollback(), you must call the .prepare() coroutine before committing.
TwoPhaseTransaction properties:
xid: Returns the two-phase transaction ID.prepare(): Prepares the transaction for commit.Standard cursors buffer the entire result set in memory. For very large queries or slow networks, use unbuffered cursors:
SSCursor: An unbuffered cursor. It fetches rows as needed, significantly reducing client-side memory usage.SSDictCursor: An unbuffered cursor that returns results as dictionaries.Limitations of Unbuffered Cursors:
fetchall() is inefficient as it still fetches all rows one by one.DBAPI used by most relational database modules. The primary difference is that you must use await when calling methods on connection and cursor objects (e.g., await conn.execute() instead of conn.execute()).By default, autocommit is set to False. This means you must manually call await conn.commit() on the Connection object to persist changes made via cur.execute().
Alternatively, you can enable automatic commits by passing autocommit=True to the aiomysql.connect() method.
A Cursor allows you to execute MySQL commands within a database session. Cursors are created via Connection.cursor(). Note that cursors created from the same connection share the same session; changes made by one cursor are immediately visible to others on that same connection.
To use a cursor, you typically follow this lifecycle: connect, create cursor, execute query, fetch results, close cursor, and close connection.
import asyncio
import aiomysql
loop = asyncio.get_event_loop()
async def test_example():
conn = await aiomysql.connect(host='127.0.0.1', port=3306,
user='root', password='',
db='mysql', loop=loop)
# create default cursor
cursor = await conn.cursor()
# execute sql query
await cursor.execute("SELECT Host, User FROM user")
# fetch all results
r = await cursor.fetchall()
# detach cursor from connection
await cursor.close()
# close connection
conn.close()
loop.run_until_complete(test_example())Instead of managing individual Connection objects, use aiomysql.create_pool() to manage a pool of connections. This is the recommended way to handle multiple database operations efficiently.
To use a pool:
await aiomysql.create_pool(...).async with pool.acquire() as conn:.async with conn.cursor() as cur:.pool.close() and wait for it to finish with await pool.wait_closed().import asyncio
import aiomysql
loop = asyncio.get_event_loop()
async def go():
pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
user='root', password='',
db='mysql', loop=loop, autocommit=False)
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT 10")
(r,) = await cur.fetchone()
assert r == 10
pool.close()
await pool.wait_closed()
loop.run_until_complete(go())Instead of using raw SQL strings, you can use aiomysql.sa to leverage SQLAlchemy's functional SQL layer as a query builder. This allows you to execute SQLAlchemy expressions (like table.insert() or table.select()) directly through an aiomysql engine. The API is designed to be familiar to users of aiopg_.
import asyncio
import sqlalchemy as sa
from aiomysql.sa import create_engine
metadata = sa.MetaData()
tbl = sa.Table(
"tbl",
metadata,
sa.Column("id", sa.Integer, primary_key=True),
sa.Column("val", sa.String(255)),
)
async def go():
engine = await create_engine(
user="root",
db="test_pymysql",
host="127.0.0.1",
password="",
)
async with engine.acquire() as conn:
async with conn.begin() as transaction:
await conn.execute(tbl.insert().values(val="abc"))
await transaction.commit()
res = await conn.execute(tbl.select())
async for row in res:
print(row.id, row.val)
engine.close()
await engine.wait_closed()
asyncio.run(go())To use aiomysql, ensure your environment meets the following requirements:
aiomysql provides an asyncio DBAPI-like interface for MySQL. It is based on PyMySQL, so it shares the same API, but you must await method calls that perform I/O. Properties remain unchanged and do not require await.
import asyncio
import aiomysql
loop = asyncio.get_event_loop()
async def test_example():
# Establish connection
conn = await aiomysql.connect(host='127.0.0.1', port=3306,
user='root', password='', db='mysql',
loop=loop)
# Create a cursor and execute a query
cur = await conn.cursor()
await cur.execute("SELECT Host,User FROM user")
# Access metadata and fetch results
print(cur.description)
r = await cur.fetchall()
print(r)
# Clean up
await cur.close()
conn.close()
loop.run_until_complete(test_example())