3. ORM 选型与配置：SQLAlchemy 2.0 异步引擎配置、声明式模型基类、连接池调优

在微服务与异步中间件集成场景中，ORM 层承担着对象关系映射、连接生命周期管理、SQL 生成与执行等核心职责。本章聚焦 SQLAlchemy 2.0 的异步引擎配置、声明式模型基类设计以及连接池参数调优，确保数据库迁移与异步任务接口调用在并发环境下稳定、高效运行。

3.1 异步引擎配置

SQLAlchemy 2.0 原生支持 asyncio 事件循环，通过 create_async_engine 工厂函数创建异步引擎。核心配置项包括数据库 URL、连接池参数、日志与超时设置。

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

ASYNC_DB_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

async_engine = create_async_engine(
    ASYNC_DB_URL,
    echo=False,                     # 生产环境关闭 SQL 日志
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_pre_ping=True,
    pool_reset_on_return='rollback',
)

AsyncSessionLocal = sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

3.2 声明式模型基类

SQLAlchemy 2.0 推荐使用 声明式映射（Declarative Mapping） 定义模型。通过自定义基类，可以统一注入公共字段、表命名约定、JSON 序列化方法以及审计时间戳。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import DateTime, func
from datetime import datetime

class Base(DeclarativeBase):
    """全局声明式基类，所有业务模型继承此类"""
    pass

class TimestampMixin:
    """混入类：自动添加创建时间与更新时间"""
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )

class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(unique=True, index=True)
    email: Mapped[str] = mapped_column(unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)

3.3 连接池调优

连接池是异步 ORM 性能的关键瓶颈。调优目标：在避免连接泄漏的前提下，最大化吞吐量，最小化等待时间。

3.3.1 核心参数调优策略

3.3.2 连接池状态监控

生产环境应暴露连接池指标，用于实时监控与动态扩缩容。

from sqlalchemy import event

@event.listens_for(async_engine.sync_engine, "connect")
def receive_connect(dbapi_connection, connection_record):
    print(f"新连接建立: {dbapi_connection}")

@event.listens_for(async_engine.sync_engine, "checkout")
def receive_checkout(dbapi_connection, connection_record, connection_proxy):
    print(f"连接被取出: 池内空闲={connection_record.info.get('pool_size')}")

# 通过 engine.pool.status() 获取当前池状态
# 示例：pool_size=10, checkedin=8, checkedout=2, overflow=0

3.4 异步引擎与连接池架构

下图展示了异步引擎、连接池、会话工厂与业务代码之间的协作关系。

3.5 异步会话使用模式

推荐使用 依赖注入 + 上下文管理器 模式管理会话生命周期，确保连接及时归还。

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# FastAPI 路由中使用
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.get(User, user_id)
    return result