中文
devexamples

SQLAlchemy Async Session Engine Pool

Drive async SQLAlchemy sessions with pooled engines, per-request session scopes, and transaction guards that always release connections.

PythonData Fetching

Code

snippet
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())

Related Examples