SQLAlchemy 异步会话连接池

示例说明

异步 ORM 的资源纪律比同步更苛刻：连接不归还会迅速耗尽池子拖垮全部请求。async with SessionFactory 的会话上下文保证 close 必然执行——连接回池；session.begin 的事务上下文把提交回滚原子化，异常即回滚。两个转账并发执行验证池在竞争下的正确调度。

技术要点

• 会话上下文确保连接必归池
• begin 上下文自动提交回滚
• pool_recycle 防长连接被服务端掐断

代码解读

引擎配置十常驻加二十溢出的池参数与五秒获取超时；1800 秒回收周期匹配数据库的空闲超时防借用死连接。transfer 的两个更新语句在同一 begin 里成为原子单元；main 的 gather 并发跑两笔反向转账后汇总余额验证净效应。dispose 收尾关闭全部池连接。

使用建议

会话中禁止跨 await 持有未提交事务过久。池大小按数据库总预算与实例数折算。溢出连接是短时弹性，不该成为常态。

代码示例

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    async_sessionmaker,
    create_async_engine,
)

engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost/app',
    pool_size=10,
    max_overflow=20,
    pool_timeout=5,
    pool_recycle=1800,
    echo=False,
)

SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def transfer(from_id: int, to_id: int, amount: int) -> None:
    async with SessionFactory() as session:
        async with session.begin():
            await session.execute(
                text('UPDATE accounts SET balance = balance - :a WHERE id = :i'),
                {'a': amount, 'i': from_id},
            )
            await session.execute(
                text('UPDATE accounts SET balance = balance + :a WHERE id = :i'),
                {'a': amount, 'i': to_id},
            )


async def report_total() -> int:
    async with SessionFactory() as session:
        result = await session.execute(text('SELECT sum(balance) FROM accounts'))
        return result.scalar_one()


async def main() -> None:
    await asyncio.gather(
        transfer(1, 2, 50),
        transfer(2, 1, 20),
    )
    print('net effect applied:', await report_total())
    await engine.dispose()


asyncio.run(main())