Flask 的世界里，表单验证首选 WTForms；FastAPI 的世界里，数据验证首选 Pydantic。

pymysql 同步 flask使用

import pymysql

# 同步连接，会阻塞当前线程
connection = pymysql.connect(
    host='localhost',
    user='root',
    password='123456',
    database='test'
)

try:
    with connection.cursor() as cursor:
        # 同步执行，等待期间线程被阻塞
        cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
        result = cursor.fetchone()
        print(result)
        
        # 事务操作
        connection.begin()
        cursor.execute("UPDATE users SET name = %s WHERE id = %s", ("newname", 1))
        connection.commit()
finally:
    connection.close()

aiomysql 异步 fastAPI使用

import asyncio
import aiomysql

async def main():
    # 异步连接，不会阻塞事件循环
    pool = await aiomysql.create_pool(
        host='localhost',
        user='root',
        password='123456',
        db='test',
        minsize=5,
        maxsize=20
    )
    # acquire 获得 
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            # 异步执行，等待期间让出控制权
            await cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
            result = await cursor.fetchone()
            print(result)
            
            # 事务操作
            await conn.begin()
            await cursor.execute("UPDATE users SET name = %s WHERE id = %s", ("newname", 1))
            await conn.commit()
    
    pool.close()
    await pool.wait_closed()

# 运行异步函数
asyncio.run(main())