异步操作mysql数据库

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine('mysql+pymysql://user:password@host/dbname', poolclass=QueuePool, pool_size=10, max_overflow=20)

import asyncio
import aiomysql

async def test_example():
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306, user='root', password='', db='mysql')
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 42;")
            value = await cur.fetchone()
            print(value)
    pool.close()
    await pool.wait_closed()

asyncio.run(test_example())

try:
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM non_existent_table;")
            result = await cur.fetchall()
except Exception as e:
    print(f"Database error: {e}")

import asyncio
import aiomysql

async def fetch_data(pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM your_table;")
            result = await cur.fetchall()
            return result

async def main():
    pool = await aiomysql.create_pool(
        host='127.0.0.1',
        port=3306,
        user='root',
        password='',
        db='your_database',
        loop=asyncio.get_event_loop()
    )
    data = await fetch_data(pool)
    print(data)
    pool.close()
    await pool.wait_closed()

asyncio.run(main())