数据库会话

说明

通过 contextvars + Starlette 中间件实现请求级数据库会话管理，写操作按业务成功与否自动 commit/rollback。核心代码位于 src/core/database.py 和 src/middleware/db_session.py。

核心架构

数据库会话管理位于 src/core/database.py 和 src/middleware/db_session.py。

contextvars 隔离

from contextvars import ContextVar

# 请求级数据库 session 上下文变量
_db_ctx: ContextVar = ContextVar('_db_ctx')

每个请求通过 contextvars 持有独立的数据库会话，协程之间互不干扰。

get_db_session()

业务代码通过 get_db_session() 获取当前请求的数据库会话：

def get_db_session():
    """获取当前请求的数据库 session（仅限请求上下文内调用）"""
    try:
        return _db_ctx.get()
    except LookupError:
        raise RuntimeError(
            "get_db_session() 只能在请求上下文内调用（缺少 db_session 中间件注入）。"
            "后台线程/启动阶段请使用 SessionLocal() 创建独立会话。"
        ) from None

温馨提示

get_db_session() 只能在请求上下文内调用。后台线程或启动阶段需要使用 SessionLocal() 创建独立会话。

会话中间件

src/middleware/db_session.py 中的中间件负责会话的创建、提交和关闭：

async def db_session_middleware(request: Request, call_next):
    """请求级 DB 会话：写操作按业务成功与否 commit/rollback，finally 关闭"""
    db = SessionLocal()
    _db_ctx.set(db)
    try:
        response = await call_next(request)
        if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
            body, consumed = await read_response_body(response)
            if consumed:
                response = rebuild_response(response, body)
                if _is_business_success(body):
                    db.commit()
                else:
                    db.rollback()
        return response
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

业务成功判断

中间件通过响应体 JSON 的 code 字段判断业务是否成功：

def _is_business_success(body: bytes) -> bool:
    """仅当响应体为 JSON 且业务码 code == 0 视为成功"""
    if not body:
        return False
    try:
        data = json.loads(body)
    except Exception:
        return False
    if not isinstance(data, dict):
        return False
    return data.get('code') == 0

独立提交

commit_independently() 函数用于审计日志等需要无论业务成功与否都持久化的场景：

def commit_independently(obj):
    """
    使用独立session提交对象，不受请求事务影响

    日志等审计数据无论业务成功或失败都应持久化，
    因此需要绕过 AOP 事务中间件的 commit/rollback 机制。
    """
    independent_session = SessionLocal()
    try:
        independent_session.add(obj)
        independent_session.commit()
    except Exception:
        independent_session.rollback()
        raise
    finally:
        independent_session.close()

使用场景

commit_independently() 创建独立会话，不受请求级事务影响。适用于操作日志、登录日志等审计数据的持久化。普通业务数据应使用 get_db_session() 获取请求级会话，由中间件统一管理事务。

会话生命周期

请求进入
 → 中间件创建 SessionLocal 实例
 → 注入 _db_ctx
 → 业务代码通过 get_db_session() 获取会话
 → 中间件根据响应码 commit/rollback
 → finally 关闭会话

中间件注册顺序

在 src/core/app.py 中，中间件按以下顺序注册：

CORS → redis → rate_limit → db_session → operation_log → logger

db_session 在 rate_limit 之后，确保限流拦截的请求不会创建数据库会话。

总结

数据库会话管理通过 contextvars 实现请求级隔离，中间件自动根据业务响应码 commit/rollback，commit_independently() 提供独立事务能力用于审计数据持久化。这种设计确保了事务边界清晰，且不会阻塞事件循环。