深入理解SQLAlchemy异步会话与PostgreSQL连接池管理

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

# 示例：将连接池大小设置为10
# DATABASE_URL 替换为您的实际数据库连接字符串
db_engine = create_async_engine(
    '<DATABASE_URL>',
    echo=False,
    future=True,
    pool_size=10  # 设置连接池大小为10
)
async_session = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        # 在此代码块内，session 是一个活跃的数据库会话
        # 可以执行数据库操作，例如：
        # await session.execute(text("SELECT 1"))
        yield session
    # 当代码执行到这里，即退出 'async with' 块时，
    # SQLAlchemy会自动关闭会话（如果需要）或将底层连接返回到连接池。
    # 因此，显式调用 await session.close() 是不必要的，甚至可能导致逻辑混乱。

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import text
import asyncio

# 1. 配置数据库引擎和连接池
# 替换为您的实际数据库连接字符串
DATABASE_URL = "postgresql+asyncpg://user:password@host:port/database"

db_engine = create_async_engine(
    DATABASE_URL,
    echo=False,  # 设置为True可以打印SQL语句，便于调试
    future=True,
    pool_size=10,  # 示例：设置连接池大小为10
    max_overflow=5, # 示例：允许额外创建5个连接
    pool_recycle=3600 # 示例：连接每小时回收一次
)

# 2. 创建异步会话工厂
async_session = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)

# 3. 定义获取会话的依赖函数（常用于依赖注入）
async def get_db_session() -> AsyncSession:
    """
    提供一个异步数据库会话，使用上下文管理器自动管理会话生命周期。
    """
    async with async_session() as session:
        yield session

# 4. 示例：如何使用获取到的会话执行操作
async def perform_database_operation():
    async for session in get_db_session():
        try:
            result = await session.execute(text("SELECT version();"))
            print(f"PostgreSQL Version: {result.scalar_one()}")

            # 示例：执行一个事务
            await session.execute(text("INSERT INTO my_table (data) VALUES ('test_data');"))
            await session.commit() # 提交事务
            print("数据插入成功并提交。")
        except Exception as e:
            await session.rollback() # 发生错误时回滚事务
            print(f"数据库操作失败: {e}")
        finally:
            # 退出 async with 块时，会话会自动关闭或连接返回到连接池
            pass

# 5. 运行示例
if __name__ == "__main__":
    asyncio.run(perform_database_operation())
    # 应用程序结束时，可以关闭引擎，释放所有连接
    # asyncio.run(db_engine.dispose()) # 如果应用程序完全退出，可以调用此方法