2024年3月9日发(作者：)

python异步操作MySQL（aiomysql）

基本的异步connection

import asyncio

from aiomysql import create_pool

loop = _event_loop()

async def go():

async with create_pool(host='127.0.0.1', port=3306,

user='root', password='',

db='mysql', loop=loop) as pool:

async with () as conn:

async with () as cur:

await e("SELECT 42;")

value = await ne()

print(value)

_until_complete(go())

连接 pool

import asyncio

import aiomysql

async def test_example(loop):

pool = await _pool(host='127.0.0.1', port=3306,

user='root', password='',

db='mysql', loop=loop)

async with e() as conn:

async with () as cur:

await e("SELECT 42;")

print(ption)

(r,) = await ne()

assert r == 42

()

await _closed()

loop = _event_loop()

_until_complete(test_example(loop))

对象关系映射SQLAlchemy - Object Relationship Mapping

import asyncio

import sqlalchemy as sa

from import create_engine

metadata = ta()

tbl = ('tbl', metadata,

('id', r, primary_key=True),

('val', (255)))

async def go(loop):

engine = await create_engine(user='root', db='test_pymysql',

host='127.0.0.1', password='', loop=loop)

async with e() as conn:

await e(().values(val='abc'))

await e(().values(val='xyz'))

async for row in e(()):

print(, )

()

await _closed()

loop = _event_loop()

_until_complete(go(loop))