第38章：【SQLAlchemy 高级篇综合实战】自研轻量数据访问框架

davidwang456

259人浏览 · 2026-08-21 10:27:29

davidwang456 · 2026-08-21 10:27:29 发布

一、项目背景

“团队 30 个微服务，每个服务都手写 SQLAlchemy 的 Session 管理、租户过滤、审计日志、读写分离——代码高度重复、各自有各自的 bug。能不能抽一个公司级的 DAO 框架，一次写对、全司复用？”

星云电商经过三年的发展，微服务数量从 3 个膨胀到 30 个。每个服务都用 SQLAlchemy，但使用方式各不相同：

服务 A 的 Session 管理用 async with + expire_on_commit=True——导致提交后访问属性报 DetachedInstanceError。
服务 B 的租户过滤依靠每个查询手动 WHERE tenant_id = ?——三个月内出现 2 次漏加导致的数据泄露。
服务 C 的审计日志用 before_flush 事件——但忘记在 bulk 操作后额外处理，批量导入的数据缺失审计。
服务 D 的读写分离用 text() 硬编码路由——无法动态切换到只读副本。

架构师算了一笔账：30 个服务 × 每人每月花 2 小时排查 ORM 相关问题 = 每月 60 人时——一年就是 720 人时（相当于 0.4 个 FT 的人力）。如果能把通用能力抽象为一个薄薄的 DAO 框架层——统一 Session 管理、全局租户过滤、自动审计、读写分离、指标埋点——不仅能消除重复代码，更能让 ORM 问题被"一次修复、全司受益"。

本章是高级篇的综合实战——在 SQLAlchemy 之上构建一个企业级轻量数据访问框架（命名为 NebulaDAO），将前 37 章的全部知识点整合进一个可交付的组件中。交付物包括：框架源码、压测报告、Runbook、迁移规范、跨角色培训材料。

二、项目设计

场景一：架构评审

架构评审会上，大师展示了 30 个微服务中共同存在的 ORM 反模式清单。C T O 问：“能不能封装一个公司级的薄层，让每个服务不用重复造轮子？”

小胖：“这不就跟 Spring Data JPA 一样——定义一个接口就能自动实现 CRUD？”

大师：“方向对，但要更轻量。Spring Data 是一个完整框架——重量级、约束多。我们要做的是一个薄适配层——不隐藏 SQLAlchemy，而是增强它。开发者仍然可以使用 SQLAlchemy 的全部 API，但通过框架自动获得：租户过滤、审计日志、读写分离、指标埋点。”

小白：“框架的接口是什么？Service 代码怎么调用？”

大师：“核心接口是一个 Repository 基类——”

# Service 层的调用方式
class OrderRepository(Repository[Order]):
    """订单仓储——自动获得租户过滤、审计、读写分离"""

    async def find_by_user(self, user_id: int) -> List[Order]:
        # Repository 自动注入的 session 已经带租户过滤
        stmt = select(Order).where(Order.user_id == user_id)
        return await self.all(stmt)

    async def create(self, order_data: dict) -> Order:
        # 创建自动触发审计事件
        order = Order(**order_data)
        self.session.add(order)
        await self.session.commit()
        return order

小胖：“技术映射：Repository = 24小时便利店（标准化的结账流程、统一的进货渠道、自动的库存管理——每家分店不用重新设计）。”

小白：“框架怎么保证所有查询都自动加 tenant_id 过滤？”

大师：“框架在 Session 创建后立即注册 do_orm_execute 事件——在框架层完成，业务层零感知。如果某个查询需要绕过——用 execution_options 标记。”

场景二：读写分离与路由策略

大师：“再看读写分离——框架内置了一个 RoutingSession，根据操作类型自动路由。”

class RoutingSession(AsyncSession):
    def get_bind(self, mapper=None, clause=None, **kw):
        if self._is_write_operation(clause):
            return write_engine  # 写入 → 主库
        return read_engine       # 查询 → 只读副本

小胖：“那如果查询需要强一致性（读刚写入的数据）——在副本上看不到怎么办？”

大师：“用 execution_options(use_master=True) 临时路由到主库。框架也支持显式声明读写类型。”

场景三：迁移规范与运维决策

大师：“框架还带了一套迁移规范：每个服务的 Alembic 迁移都必须通过 CI 检查——alembic check + upgrade → downgrade → upgrade 循环。不能在高峰期跑 DDL。所有迁移必须有回滚脚本。”

小胖：“技术映射：迁移规范 = 交通规则（红灯停、绿灯行、变道要打灯）——没有规则的交通会拥堵（CI 失败），有规则才能高效运转。”

三、项目实战

3.1 框架骨架——核心模块

"""nebula_dao/__init__.py —— NebulaDAO 框架入口"""

# 框架目录结构
NEBULA_DAO_STRUCTURE = """
nebula_dao/
├── __init__.py              # 框架入口
├── config.py                # 配置管理（连接池参数、SQL 日志阈值等）
├── engine.py                # 引擎工厂（同步/异步 + 连接池配置）
├── session.py               # Session/RoutingSession/AsyncSession 管理器
├── repository.py            # Repository 基类（CRUD + 查询模板）
├── mixins.py                # TenantMixin, SoftDeleteMixin, TimestampMixin
├── events.py                # 全局事件注册（租户过滤、审计、慢 SQL、指标）
├── audit.py                 # 审计日志模型与写入逻辑
├── metrics.py               # Prometheus 指标采集与暴露
├── migrations/              # Alembic 迁移规范与 CI 检查脚本
└── runbook.md               # 生产故障处理手册
"""
print(NEBULA_DAO_STRUCTURE)

3.2 配置管理——统一入口

# ===== nebula_dao/config.py =====
"""
import os
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class DatabaseConfig:
    host: str = os.getenv("DB_HOST", "localhost")
    port: int = int(os.getenv("DB_PORT", "5432"))
    database: str = os.getenv("DB_NAME", "order_center")
    user: str = os.getenv("DB_USER", "nebula")
    password: str = field(default_factory=lambda: os.getenv("DB_PASSWORD", ""))

    # 只读副本（读写分离）
    read_host: Optional[str] = os.getenv("DB_READ_HOST", None)
    read_port: int = int(os.getenv("DB_READ_PORT", "5432"))

    @property
    def master_url(self) -> str:
        return f"postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"

    @property
    def replica_url(self) -> str:
        if self.read_host:
            return f"postgresql+asyncpg://{self.user}:{self.password}@{self.read_host}:{self.read_port}/{self.database}"
        return self.master_url  # 无只读副本时回退到主库

@dataclass
class PoolConfig:
    pool_size: int = 5
    max_overflow: int = 10
    pool_timeout: int = 30
    pool_recycle: int = 3600
    pool_pre_ping: bool = True
    echo: bool = False

@dataclass
class FrameworkConfig:
    \"\"\"框架全局配置\"\"\"
    database: DatabaseConfig = field(default_factory=DatabaseConfig)
    pool: PoolConfig = field(default_factory=PoolConfig)
    slow_sql_threshold_ms: int = 300
    enable_audit: bool = True
    enable_tenant_filter: bool = True
    enable_soft_delete_filter: bool = True
    enable_metrics: bool = True
    enable_read_write_split: bool = True
"""

print("=== 配置管理 ===")
print("  支持: DatabaseConfig + PoolConfig + FrameworkConfig")
print("  配置来源: 环境变量 / K8s Secret / Vault")

3.3 引擎工厂——同步与异步统一创建

# ===== nebula_dao/engine.py =====
"""
from sqlalchemy import create_engine, event
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.pool import QueuePool, NullPool, AsyncAdaptedQueuePool
from nebula_dao.config import FrameworkConfig
import logging

logger = logging.getLogger("nebula_dao")

class EngineFactory:
    '''引擎工厂——统一创建同步/异步引擎，注入连接池事件'''

    @staticmethod
    def create_async_engine(config: FrameworkConfig) -> AsyncEngine:
        engine = create_async_engine(
            config.database.master_url,
            pool_size=config.pool.pool_size,
            max_overflow=config.pool.max_overflow,
            pool_timeout=config.pool.pool_timeout,
            pool_recycle=config.pool.pool_recycle,
            pool_pre_ping=config.pool.pool_pre_ping,
            echo=config.pool.echo,
        )
        EngineFactory._register_pool_events(engine)
        return engine

    @staticmethod
    def create_sync_engine(config: FrameworkConfig):
        engine = create_engine(
            config.database.master_url,
            pool_size=config.pool.pool_size,
            max_overflow=config.pool.max_overflow,
            pool_timeout=config.pool.pool_timeout,
            pool_recycle=config.pool.pool_recycle,
            pool_pre_ping=config.pool.pool_pre_ping,
            echo=config.pool.echo,
        )
        EngineFactory._register_pool_events(engine)
        return engine

    @staticmethod
    def _register_pool_events(engine):
        '''注册连接池事件——日志 + 指标采集'''
        @event.listens_for(engine, "checkout")
        def on_checkout(dbapi_conn, conn_record, conn_proxy):
            logger.debug(f"连接 checkout: pool_size={engine.pool.size()} overflow={engine.pool.overflow()}")

        @event.listens_for(engine, "checkin")
        def on_checkin(dbapi_conn, conn_record):
            logger.debug(f"连接 checkin: checked_out={engine.pool.checkedout()}")

        @event.listens_for(engine, "close")
        def on_close(dbapi_conn, conn_record):
            logger.debug("连接关闭")
"""

print("=== 引擎工厂 ===")
print("  提供: create_async_engine / create_sync_engine")
print("  自动注入: 连接池事件监控")

3.4 Session管理器——统一生命周期 + 读写分离路由

# ===== nebula_dao/session.py =====
"""
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.sql import Select, Update, Delete, Insert
from nebula_dao.config import FrameworkConfig
from nebula_dao.events import register_global_events
from contextlib import asynccontextmanager, contextmanager
from typing import AsyncGenerator, Generator
import contextvars

# 租户上下文
current_tenant: contextvars.ContextVar[str] = contextvars.ContextVar('tenant_id', default='')

class RoutingSession(AsyncSession):
    '''读写分离路由——根据 SQL 类型自动选择主库/只读副本'''

    def __init__(self, write_bind, read_bind, **kwargs):
        self._write_bind = write_bind
        self._read_bind = read_bind
        super().__init__(bind=write_bind, **kwargs)

    def get_bind(self, mapper=None, clause=None, **kw):
        if self._is_write(clause):
            return self._write_bind
        return self._read_bind

    def _is_write(self, clause) -> bool:
        if clause is None:
            return self._force_write
        return isinstance(clause, (Update, Delete, Insert))

class SessionManager:
    '''Session 管理器——创建、注入事件、提供 Depends 函数'''

    def __init__(self, config: FrameworkConfig, write_engine, read_engine=None):
        self.config = config
        self.write_engine = write_engine
        self.read_engine = read_engine or write_engine

        if self.read_engine is not self.write_engine:
            self.Factory = async_sessionmaker(
                class_=lambda **kw: RoutingSession(write_engine, read_engine, **kw),
                expire_on_commit=False,
            )
        else:
            self.Factory = async_sessionmaker(
                bind=write_engine, class_=AsyncSession,
                expire_on_commit=False,
            )

        # 注册全局事件
        register_global_events(config)

    async def get_session(self) -> AsyncGenerator[AsyncSession, None]:
        '''FastAPI Depends 使用的 Session 依赖'''
        async with self.Factory() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise

    @asynccontextmanager
    async def begin(self) -> AsyncGenerator[AsyncSession, None]:
        '''手动管理的事务上下文'''
        async with self.Factory() as session:
            async with session.begin():
                yield session
"""

print("=== Session 管理器 ===")
print("  提供: SessionManager.get_session() → FastAPI Depends")
print("  内置: RoutingSession 读写分离路由")
print("  自动: 全局事件注册（租户过滤、审计、慢SQL）")

3.5 Repository 基类——CRUD 模板

# ===== nebula_dao/repository.py =====
"""
from typing import TypeVar, Generic, List, Optional, Type
from sqlalchemy import select, func, delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import Select

T = TypeVar('T')

class Repository(Generic[T]):
    '''泛型 Repository 基类——提供 CRUD 模板方法'''

    entity_class: Type[T]

    def __init__(self, session: AsyncSession):
        self.session = session

    async def get(self, id: int) -> Optional[T]:
        '''按主键获取（自动受租户过滤保护）'''
        stmt = select(self.entity_class).where(self.entity_class.id == id)
        result = await self.session.execute(stmt)
        return result.scalars().first()

    async def all(self, stmt: Select = None) -> List[T]:
        '''获取全部（自动受租户过滤保护）'''
        if stmt is None:
            stmt = select(self.entity_class)
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def count(self, stmt: Select = None) -> int:
        '''计数查询'''
        if stmt is None:
            stmt = select(self.entity_class)
        count_stmt = select(func.count()).select_from(stmt.subquery())
        result = await self.session.execute(count_stmt)
        return result.scalar_one()

    async def save(self, entity: T) -> T:
        '''保存实体（新增或更新）'''
        self.session.add(entity)
        await self.session.flush()
        return entity

    async def save_all(self, entities: List[T]) -> List[T]:
        '''批量保存'''
        self.session.add_all(entities)
        await self.session.flush()
        return entities

    async def soft_delete(self, id: int) -> bool:
        '''软删除（如果有 deleted_at 字段）'''
        entity = await self.get(id)
        if entity is None:
            return False
        if hasattr(entity, 'deleted_at'):
            from datetime import datetime
            entity.deleted_at = datetime.now()
            self.session.add(entity)
            await self.session.flush()
            return True
        await self.session.delete(entity)
        await self.session.flush()
        return True

    async def execute_query(self, stmt) -> List[T]:
        '''执行自定义查询'''
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    def query(self) -> Select:
        '''获取基础查询（带所有默认过滤）'''
        return select(self.entity_class)

    def skip_tenant(self) -> Select:
        '''绕过租户过滤的查询（仅管理端使用）'''
        return select(self.entity_class).execution_options(skip_tenant_filter=True)

    def include_deleted(self) -> Select:
        '''包含已删除记录的查询'''
        return select(self.entity_class).execution_options(include_deleted=True)
"""

print("=== Repository 基类 ===")
print("  泛型设计: Repository[Order] → 自动 CRUD")
print("  内置: get / all / count / save / save_all / soft_delete")
print("  自动: 所有查询继承全局过滤（租户 + 软删除）")

3.6 Mixins——可复用的模型增强

# ===== nebula_dao/mixins.py =====
"""
from sqlalchemy import String, DateTime, Index, func
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from typing import Optional

class TenantMixin:
    '''多租户 Mixin'''
    tenant_id: Mapped[str] = mapped_column(String(50), nullable=False, index=True)

class SoftDeleteMixin:
    '''软删除 Mixin'''
    deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, default=None, index=True)

class TimestampMixin:
    '''时间戳 Mixin'''
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=func.now(), nullable=True)

class AuditMixin:
    '''审计 Mixin——标记哪些模型需要审计'''
    __audit_enabled__: bool = True
"""

print("=== Mixins ===")
print("  TenantMixin / SoftDeleteMixin / TimestampMixin / AuditMixin")
print("  新模型继承 Mixins 即可获得相应能力")

3.7 全局事件——租户/审计/慢SQL/指标一体化

# ===== nebula_dao/events.py =====
"""
import time, json, logging
from sqlalchemy import event, inspect
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import with_loader_criteria
from nebula_dao.config import FrameworkConfig
from nebula_dao.mixins import TenantMixin, SoftDeleteMixin, AuditMixin
from nebula_dao.audit import AuditLog
from nebula_dao.session import current_tenant

logger = logging.getLogger("nebula_dao.events")

def register_global_events(config: FrameworkConfig):
    '''一次性注册所有全局事件'''

    # ---- 1. 租户 + 软删除全局过滤 ----
    @event.listens_for(AsyncSession, "do_orm_execute")
    def apply_filters(execute_state):
        if not execute_state.is_select:
            return
        tid = current_tenant.get()
        skip_tenant = execute_state.execution_options.get("skip_tenant_filter")
        include_deleted = execute_state.execution_options.get("include_deleted")
        filters = []
        if config.enable_tenant_filter and tid and not skip_tenant:
            filters.append(with_loader_criteria(TenantMixin, lambda cls: cls.tenant_id == tid, include_aliases=True))
        if config.enable_soft_delete_filter and not include_deleted:
            filters.append(with_loader_criteria(SoftDeleteMixin, lambda cls: cls.deleted_at == None, include_aliases=True))
        if filters:
            execute_state.statement = execute_state.statement.options(*filters)

    # ---- 2. 审计日志 ----
    if config.enable_audit:
        @event.listens_for(AsyncSession, "before_flush")
        def audit_on_flush(session, flush_context, instances):
            for obj in session.dirty:
                if isinstance(obj, AuditLog):
                    continue
                if not getattr(type(obj), '__audit_enabled__', False):
                    continue
                insp = inspect(obj)
                for attr in insp.attrs:
                    if not hasattr(attr, 'history'):
                        continue
                    hist = attr.history
                    if not hist.has_changes():
                        continue
                    old_val = str(hist.deleted[0]) if hist.deleted else None
                    new_val = str(hist.added[0]) if hist.added else None
                    audit_entry = AuditLog(
                        entity_type=type(obj).__name__,
                        entity_id=getattr(obj, 'id', None),
                        field_name=attr.key,
                        old_value=old_val,
                        new_value=new_val,
                        tenant_id=getattr(obj, 'tenant_id', ''),
                    )
                    session.add(audit_entry)

    # ---- 3. 慢 SQL 检测 ----
    def _register_slow_sql(engine):
        @event.listens_for(engine.sync_engine, "before_cursor_execute")
        def start_timer(conn, cursor, statement, params, ctx, em):
            conn.info["sql_start"] = time.monotonic()

        @event.listens_for(engine.sync_engine, "after_cursor_execute")
        def log_slow_sql(conn, cursor, statement, params, ctx, em):
            start = conn.info.pop("sql_start", None)
            if start and (time.monotonic() - start) * 1000 > config.slow_sql_threshold_ms:
                logger.warning(json.dumps({
                    "event": "slow_sql",
                    "elapsed_ms": round((time.monotonic() - start) * 1000, 2),
                    "statement": statement[:300],
                    "tenant_id": current_tenant.get(""),
                }, ensure_ascii=False))
"""

print("=== 全局事件 ===")
print("  一体化注册: 租户过滤 + 审计日志 + 慢SQL检测")
print("  通过 FrameworkConfig 控制启用/禁用")

3.8 使用框架——Service 层示例

# ===== 业务代码示例 =====
"""
from nebula_dao.repository import Repository
from nebula_dao.session import SessionManager, current_tenant
from nebula_dao.mixins import TenantMixin, SoftDeleteMixin, TimestampMixin
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Numeric, ForeignKey, select
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends

# ---- 1. 模型定义（继承 Mixins 即可获得能力） ----
class Base(DeclarativeBase):
    pass

class Order(Base, TenantMixin, SoftDeleteMixin, TimestampMixin):
    __tablename__ = "orders"
    __audit_enabled__ = True
    id: Mapped[int] = mapped_column(primary_key=True)
    order_no: Mapped[str] = mapped_column(String(32))
    user_id: Mapped[int] = mapped_column(Integer)
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2))
    status: Mapped[str] = mapped_column(String(20), default="pending")

# ---- 2. Repository 定义 ----
class OrderRepository(Repository[Order]):
    entity_class = Order

    async def find_by_user(self, user_id: int):
        stmt = self.query().where(Order.user_id == user_id)
        return await self.all(stmt)

    async def find_large_orders(self, min_amount: float):
        stmt = self.query().where(Order.total_amount >= min_amount)
        return await self.all(stmt)

# ---- 3. Service 层 ----
class OrderService:
    def __init__(self, session: AsyncSession):
        self.repo = OrderRepository(session)

    async def create_order(self, user_id: int, amount: float):
        order = Order(
            order_no=f"ORD-{uuid.uuid4().hex[:8].upper()}",
            user_id=user_id,
            total_amount=amount,
            tenant_id=current_tenant.get(),
        )
        return await self.repo.save(order)

    async def get_user_orders(self, user_id: int):
        return await self.repo.find_by_user(user_id)

# ---- 4. FastAPI 端点 ----
sm = SessionManager(config, write_engine, read_engine)

@app.post("/orders")
async def create_order(
    req: OrderCreateRequest,
    session: AsyncSession = Depends(sm.get_session),
    x_tenant: str = Header(alias="X-Tenant-ID"),
):
    current_tenant.set(x_tenant)
    svc = OrderService(session)
    order = await svc.create_order(req.user_id, req.amount)
    return order
"""

print("=== 使用示例 ===")
print("  Model → 继承 Mixins (TenantMixin + SoftDeleteMixin + TimestampMixin)")
print("  Repository → 继承 Repository[Order]")
print("  Service → 通过 Depends(sm.get_session) 注入 Session")
print("  ✅ 自动: 租户过滤 + 审计日志 + 软删除 + 慢SQL告警")

3.9 指标采集与 Prometheus 暴露

# ===== nebula_dao/metrics.py =====
"""
from prometheus_client import Histogram, Counter, Gauge, generate_latest, CollectorRegistry
import time
from sqlalchemy import event

_registry = CollectorRegistry(auto_describe=True)

sql_duration = Histogram(
    'nebula_sql_duration_seconds', 'SQL duration',
    ['operation'], registry=_registry,
)

sql_errors = Counter(
    'nebula_sql_errors_total', 'SQL errors',
    ['error_type'], registry=_registry,
)

pool_checked_out = Gauge(
    'nebula_pool_checked_out', 'Pool checked out',
    registry=_registry,
)

def register_metrics_collector(engine):
    @event.listens_for(engine.sync_engine, "before_cursor_execute")
    def _start(conn, cursor, statement, params, ctx, em):
        conn.info["m_start"] = time.monotonic()

    @event.listens_for(engine.sync_engine, "after_cursor_execute")
    def _record(conn, cursor, statement, params, ctx, em):
        s = conn.info.pop("m_start", None)
        if s:
            op = statement.strip().split()[0][:10]
            sql_duration.labels(operation=op).observe(time.monotonic() - s)

    @event.listens_for(engine.sync_engine, "handle_error")
    def _error(ctx):
        sql_errors.labels(error_type=type(ctx.original_exception).__name__).inc()

def get_metrics_response():
    from fastapi.responses import Response
    return Response(generate_latest(_registry), media_type="text/plain")
"""

print("=== 指标采集 ===")
print("  内置指标: sql_duration, sql_errors, pool_checked_out")
print("  暴露端点: GET /metrics → Prometheus 拉取")

3.10 迁移规范 + CI 检查

# ===== nebula_dao/migrations/ci_check.py =====
"""
#!/usr/bin/env python
import sys
import subprocess

def check_alembic_chain():
    '''检查 Alembic 迁移链是否完整'''
    result = subprocess.run(["alembic", "heads"], capture_output=True, text=True)
    heads = [h.strip() for h in result.stdout.strip().split("\\n") if h.strip()]
    if len(heads) > 1:
        print(f"错误：发现多个 head: {heads}")
        sys.exit(1)
    print(f"迁移链正常: {heads[0]}")

def check_model_migration_sync():
    '''检查模型与迁移是否同步（alembic check）'''
    result = subprocess.run(["alembic", "check"], capture_output=True, text=True)
    if result.returncode != 0:
        print(f"错误：模型与迁移不同步:\\n{result.stdout}")
        sys.exit(1)
    print("模型与迁移同步")

def check_upgrade_downgrade_cycle():
    '''测试 upgrade → downgrade → upgrade 循环'''
    subprocess.run(["alembic", "downgrade", "base"], check=True)
    subprocess.run(["alembic", "upgrade", "head"], check=True)
    print("upgrade/downgrade 循环测试通过")

def check_no_dangerous_ddl():
    '''检查迁移脚本中是否有危险 DDL 操作'''
    import os, re
    versions_dir = os.path.join(os.path.dirname(__file__), "versions")
    dangerous_patterns = [
        (r"ALTER.*TYPE", "ALTER COLUMN TYPE（全表重写）"),
        (r"ADD COLUMN.*NOT NULL", "ADD COLUMN NOT NULL（可能锁表）"),
        (r"DROP TABLE", "DROP TABLE（不可逆操作）"),
        (r"TRUNCATE", "TRUNCATE（不可逆操作）"),
    ]
    for fname in os.listdir(versions_dir):
        if fname.endswith(".py"):
            with open(os.path.join(versions_dir, fname)) as f:
                content = f.read()
            for pattern, desc in dangerous_patterns:
                if re.search(pattern, content, re.IGNORECASE):
                    print(f"警告：{fname} 包含危险操作: {desc}")
    print("DDL 安全检查完成")

if __name__ == "__main__":
    check_alembic_chain()
    check_model_migration_sync()
    check_upgrade_downgrade_cycle()
    check_no_dangerous_ddl()
"""

print("=== 迁移 CI 检查 ===")
print("  自动检查: 多 head / 模型同步 / upgrade-downgrade / 危险 DDL")

3.11 跨角色培训材料

# ===== 开发人员一页纸 =====
DEV_ONEPAGER = """
╔══════════════════════════════════════════════════════════╗
║            NebulaDAO 开发人员一页纸                        ║
╠══════════════════════════════════════════════════════════╣
║ 1. 创建模型：                                            ║
║    class Order(Base, TenantMixin, SoftDeleteMixin,       ║
║                TimestampMixin):                          ║
║        __audit_enabled__ = True                          ║
║                                                          ║
║ 2. 创建 Repository：                                     ║
║    class OrderRepo(Repository[Order]):                   ║
║        entity_class = Order                              ║
║                                                          ║
║ 3. 创建 Service：                                        ║
║    class OrderService:                                   ║
║        def __init__(self, session): self.repo = ...      ║
║                                                          ║
║ 4. FastAPI 端点：                                        ║
║    session: AsyncSession = Depends(sm.get_session)       ║
║                                                          ║
║ 5. 自动获得的能力：                                      ║
║    ✅ 租户隔离 ✅ 软删除 ✅ 审计日志 ✅ 慢SQL告警         ║
║    ✅ 读写分离 ✅ Prometheus指标                         ║
║                                                          ║
║ 6. 变更 Schema 时：                                      ║
║    alembic revision --autogenerate -m "变更描述"         ║
║    → 检查生成的 migration → 提交 PR → CI 自动验证        ║
╚══════════════════════════════════════════════════════════╝
"""

# ===== 测试人员一页纸 =====
TEST_ONEPAGER = """
╔══════════════════════════════════════════════════════════╗
║            NebulaDAO 测试人员一页纸                        ║
╠══════════════════════════════════════════════════════════╣
║ 必须覆盖的测试用例：                                      ║
║ 1. 租户隔离测试                                          ║
║    - A 创建的数据，B 查询不到                             ║
║    - 管理端 (skip_tenant_filter=True) 可跨租户查看        ║
║                                                          ║
║ 2. 软删除测试                                            ║
║    - 软删除后默认查询看不到                               ║
║    - 使用 include_deleted=True 可以看到                   ║
║                                                          ║
║ 3. 审计日志测试                                          ║
║    - 修改实体属性 → 审计日志中有记录                       ║
║    - rollback 后审计日志不落库                            ║
║                                                          ║
║ 4. N+1 检测测试                                          ║
║    - 列表页 SQL 数量 ≤ 2（主查询 + selectinload）         ║
║    - 使用 lazy='raise_on_sql' 暴露问题                    ║
║                                                          ║
║ 5. 迁移完整性测试                                        ║
║    - CI 中执行 upgrade → downgrade → upgrade 循环         ║
║    - 验证 alembic check 通过                              ║
╚══════════════════════════════════════════════════════════╝
"""

# ===== 运维人员一页纸 =====
OPS_ONEPAGER = """
╔══════════════════════════════════════════════════════════╗
║            NebulaDAO 运维人员一页纸                        ║
╠══════════════════════════════════════════════════════════╣
║ 监控关键指标 (Prometheus)：                               ║
║ 1. nebula_pool_checked_out  > 80% pool_size → 告警       ║
║ 2. nebula_sql_duration_seconds P99 > 500ms → 告警        ║
║ 3. nebula_sql_errors_total  > 0.1% → 告警                ║
║                                                          ║
║ 迁移执行规范：                                            ║
║ 1. 只允许在低峰期 (凌晨 2-4 点) 执行                      ║
║ 2. 执行前检查 pg_stat_activity 是否有长时间事务            ║
║ 3. 设置 lock_timeout = '5s'                              ║
║ 4. 准备回滚预案：alembic downgrade -1                     ║
║                                                          ║
║ 连接池参数：                                              ║
║ pool_size = 5, max_overflow = 10                         ║
║ 每 Worker 最大连接 = pool_size + max_overflow = 15        ║
║ 需确保 总连接数 ≤ 数据库 max_connections × 80%            ║
║                                                          ║
║ 常用运维命令：                                            ║
║ /health/pool → 查看当前连接池状态                         ║
║ /health/recycle → 触发连接池重建                          ║
║ /metrics → Prometheus 指标                               ║
╚══════════════════════════════════════════════════════════╝
"""

print("=== 跨角色培训材料 ===")
print(DEV_ONEPAGER)
print(TEST_ONEPAGER)
print(OPS_ONEPAGER)

3.12 压测与容量报告模板

# ===== 压测报告模板 =====
LOAD_TEST_REPORT = """
╔══════════════════════════════════════════════════════════╗
║           NebulaDAO 压测报告                               ║
╠══════════════════════════════════════════════════════════╣
║ 环境信息：                                                ║
║   数据库: PostgreSQL 16, 4 vCPU, 16GB RAM                 ║
║   应用: 10 Workers, 每个 pool_size=5, max_overflow=10     ║
║   压测工具: locust, 30 并发用户, 10 分钟预热              ║
║                                                          ║
║ 测试场景                   │ QPS   │ P50   │ P99   │ 错误率║
║ ──────────────────────────────────────────────────────────║
║ GET /orders (20条/页)      │ 450   │ 18ms  │ 52ms  │ 0%   ║
║ GET /orders?user_id=1      │ 800   │ 8ms   │ 22ms  │ 0%   ║
║ POST /orders (创建订单)    │ 320   │ 25ms  │ 85ms  │ 0%   ║
║ PUT /orders/:id/status     │ 550   │ 12ms  │ 35ms  │ 0%   ║
║ GET /orders/report (聚合)  │ 120   │ 45ms  │ 180ms │ 0%   ║
║                                                          ║
║ 连接池状态：                                              ║
║   峰值 checkout: 8/15                                     ║
║   峰值 overflow: 2                                        ║
║   超时次数: 0                                             ║
║                                                          ║
║ SQL 性能：                                                ║
║   列表页 SQL 数: 2 (主查询 + selectinload)                ║
║   N+1 检测: 无                                           ║
║   慢 SQL (>300ms): 3 条 (报表聚合查询)                    ║
║                                                          ║
║ 容量规划建议：                                            ║
║   当前: 10 Workers × 15 连接 = 150 连接                   ║
║   DB max_connections: 450                                 ║
║   安全上限: 450 × 80% = 360 连接                          ║
║   最多 Workers: 360 / 15 = 24 Workers                     ║
║   建议扩容时: 先加只读副本分担查询                         ║
╚══════════════════════════════════════════════════════════╝
"""

print("=== 压测报告 ===")
print(LOAD_TEST_REPORT)

可能遇到的坑及解决方法

框架与业务代码的循环依赖

现象：models/order.py import Repository，而 repository.py import Order——循环依赖。
解决：Repository 使用 entity_class 类属性延迟绑定——在 __init_subclass__() 中自动关联。

多个 SessionManager 实例的事件重复注册

现象：创建了两个 SessionManager 实例，事件的监听器被注册两次——审计日志双写。
解决：使用单例模式——SessionManager 全局唯一，或使用 event.contains() 检查是否已注册。

框架版本升级导致的不兼容

现象：框架依赖的 SQLAlchemy 版本升级了，框架内部调用的 API 变化。
解决：框架在 pyproject.toml 中固定 SQLAlchemy 版本范围："sqlalchemy>=2.0.20,<2.1"。

测试验证

# tests/test_ch38_framework.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import String, Integer, select, text

@pytest.mark.asyncio
async def test_repository_crud():
    """验证 Repository 基类的 CRUD 操作"""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
    from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

    class Base(DeclarativeBase):
        pass
    class T(Base):
        __tablename__ = "t"
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str] = mapped_column(String(50))

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    Factory = async_sessionmaker(engine, expire_on_commit=False)

    # 模拟 Repository
    class MockRepo:
        entity_class = T
        def __init__(self, session):
            self.session = session
        async def all(self):
            r = await self.session.execute(select(T))
            return list(r.scalars().all())
        async def save(self, entity):
            self.session.add(entity)
            await self.session.flush()
            return entity

    async with Factory() as s:
        repo = MockRepo(s)
        await repo.save(T(name="test"))
        all_items = await repo.all()
        assert len(all_items) == 1
        assert all_items[0].name == "test"
        await s.commit()

    await engine.dispose()

@pytest.mark.asyncio
async def test_routing_session_routes_correctly():
    """验证读写分离路由"""
    # 简化的测试验证路由逻辑
    from sqlalchemy import Update, Select

    assert isinstance(select(text("1")), Select)
    # 此测试验证概念而非完整实现
    assert True

四、项目总结

框架能力矩阵

能力	实现方式	开发感知	运维感知
租户隔离	`do_orm_execute` + `with_loader_criteria`	零（自动注入）	无
软删除过滤	同上	零	无
审计日志	`before_flush` 事件	标注 `__audit_enabled__ = True`	可查询审计表
读写分离	`RoutingSession.get_bind()`	零（默认只读路由到副本）	需配置副本
慢 SQL 告警	`after_cursor_execute`	零	Prometheus 告警
连接池监控	`checkout`/`checkin` 事件	零	指标面板
迁移规范	CI 检查脚本	需手动创建 migration	执行窗口

适用场景

新微服务启动：使用框架模板快速搭建 ORM 层——5 分钟获得全部能力。
老服务迁移：逐步将老服务的 ORM 代码迁移到框架——一次迁移、永久受益。
多团队协作：框架提供统一的 Repository 基类和事件注册——减少代码审查争议。

不适用场景：

极简单的 CLI 脚本（只有 2 张表）——框架的抽象成本高于收益。
不使用 ORM 的场景（如纯 Core 的 ETL 管道）——框架面向 ORM 场景。

注意事项

框架是薄层——不要隐藏 SQLAlchemy。开发者仍可使用 SQLAlchemy 的全部 API 直接操作。
框架版本与 SQLAlchemy 版本需对应——通过 pyproject.toml 声明依赖范围。
不要在框架的全局事件中做高开销操作——事件中的代码对每条 SQL 都会执行。

常见踩坑经验

案例 1：框架升级后业务代码中 session.execute(raw_text) 被全局事件意外拦截

现象：手写的 text() 语句被 do_orm_execute 中的过滤器拦截——误加 WHERE 条件。
修复：在不需要全局过滤的查询上加 .execution_options(skip_tenant_filter=True)。框架文档中明确说明哪些操作需要手动标记。

案例 2：Repository 的 all(stmt) 返回类型与业务预期不一致

现象：await repo.all(stmt) 在无结果时返回空列表，但业务期望 None（用于区分"无数据"与"错误"）。
修复：框架保持一致性——all() 永远返回列表（空列表表示无数据）。需要 None 的场景用 repo.get(id)。

案例 3：SessionManager 在多 Worker 共享下重复注册全局事件

现象：多个 Worker 进程各自创建 SessionManager——事件累加注册。
根因：@event.listens_for 在每次 SessionManager.__init__ 时都注册一次。
修复：检查 event.contains(AsyncSession, "do_orm_execute", apply_filters) 后再决定是否注册。

思考题

框架中的 Repository 基类提供了 get(id) 方法——它使用 select(Model).where(Model.id == id) 而非 session.get(Model, id)。两者的区别是什么？框架选择前者是为了兼容全局过滤——session.get() 不走 do_orm_execute，无法进行租户隔离。这种设计是正确选择还是过度设计？有没有折中方案？
框架要求所有模型继承 TenantMixin 和 SoftDeleteMixin。但如果某个表（如 sys_config 全局配置表）不需要租户隔离——它不属于任何租户——应该如何设计？强制这类表继承 Mixin 但把 tenant_id 设为固定值吗？这在实际场景中会引发什么问题？