数据库会话

说明

项目通过 Flask 的 g 对象和 before_request / after_request 钩子实现请求级数据库会话管理，写操作按业务成功与否自动 commit/rollback。核心代码位于 src/core/database.py 和 src/middleware/transaction.py。

核心架构

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

SQLAlchemy + Flask-SQLAlchemy

项目使用 Flask-SQLAlchemy 扩展，通过 db 对象管理数据库会话：

# src/core/database.py
from flask_sqlalchemy import SQLAlchemy

# 创建 Flask-SQLAlchemy 实例
db = SQLAlchemy()

在 src/app.py 的 create_app() 中初始化：

def create_app():
    app = Flask(__name__)
    # 配置数据库 URI
    app.config['SQLALCHEMY_DATABASE_URI'] = DB_URL
    # 初始化 SQLAlchemy
    db.init_app(app)
    return app

业务代码通过 db.session 获取请求级会话：

from src.core.database import db

# 添加记录
db.session.add(obj)
db.session.flush()  # 发送 SQL，获取自增 ID，不提交

# 查询
result = db.session.query(Model).filter(...).all()

Flask g 对象

Flask 的 g 对象是请求级别的全局变量，每个请求都有独立的 g 对象：

from flask import g

# 在请求开始时设置
g.user_id = 1

# 在请求处理过程中获取
user_id = g.user_id

会话中间件

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

from flask import request, g
from src.core.database import db

def register_transaction(app):
    """注册 AOP 事务管理中间件"""

    @app.before_request
    def _transaction_begin():
        """请求开始前：标记事务活跃"""
        g.transaction_active = True

    @app.after_request
    def _transaction_commit(response):
        """请求结束后：根据响应状态决定提交或回滚"""
        if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
            if response.status_code == 200:
                try:
                    body = response.get_json()
                    if body and body.get('code') == 0:
                        # 业务成功：提交事务
                        db.session.commit()
                    else:
                        # 业务失败：回滚事务
                        db.session.rollback()
                except Exception:
                    db.session.rollback()
            else:
                # HTTP 非 200：回滚事务
                db.session.rollback()
        return response

    @app.teardown_request
    def _transaction_rollback(exception):
        """请求异常时：回滚事务"""
        if exception:
            db.session.rollback()
        # 移除会话
        db.session.remove()

业务成功判断

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

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

独立提交

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

from src.core.database import db

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

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

使用场景

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

会话生命周期

请求进入
 → before_request 标记事务活跃
 → 业务代码通过 db.session 获取会话
 → after_request 根据响应码 commit/rollback
 → teardown_request 异常时 rollback 并移除会话

中间件注册顺序

在 src/app.py 的 create_app() 中，按以下顺序初始化：

def create_app():
    app = Flask(__name__)

    # 1. 加载配置（数据库、Redis 等）
    app.config.from_object(DatabaseConfig)

    # 2. 配置 CORS 跨域
    CORS(app, ...)

    # 3. 可信代理配置（ProxyFix）
    app.wsgi_app = ProxyFix(app.wsgi_app, ...)

    # 4. 注册操作日志中间件（须在事务拦截器之前）
    register_operation_log(app)

    # 5. 注册 AOP 事务拦截器
    register_transaction(app)

    # 6. 注册 API v1 路由
    register_v1(app)

    # 7. 初始化扩展（数据库、Redis、调度器等）
    register_extends(app)

    # 8. 注册中间件（登录校验、演示模式、异常处理）
    register_middleware(app)

    # 9. 注入依赖（权限提供方、操作日志写入器）
    _inject_dependencies(app)

    return app

register_operation_log 在 register_transaction 之前，确保操作日志在事务提交前记录。

BaseDB.save() 方法

BaseDB 基类提供的 save() 方法只执行 flush，不执行 commit：

from src.core.database import db
from src.common.base_model import BaseModel

class BaseDB:
    def save(self):
        """保存模型到数据库（仅 flush，不 commit）"""
        db.session.add(self)
        db.session.flush()
        return self

    def delete(self):
        """软删除模型（仅 flush，不 commit）"""
        self.is_delete = 1
        db.session.flush()

flush 的作用：

1. 将 SQL 语句发送到数据库执行
2. 获取自增 ID 等数据库生成的值
3. 但不提交事务，事务由中间件统一管理

注意事项

重要提示

1. 不要手动调用 db.session.commit()：事务由中间件统一管理，手动 commit 可能导致事务提前提交
2. 不要手动调用 db.session.rollback()：除非在 try-except 块中需要手动回滚
3. 只读操作使用 GET 方法：GET 请求不会触发事务处理
4. 写操作必须返回 R 对象：中间件通过 code 字段判断是否需要提交

总结

数据库会话管理通过 Flask 的 g 对象和 before_request / after_request 钩子实现请求级隔离，中间件自动根据业务响应码 commit/rollback，commit_independently() 提供独立事务能力用于审计数据持久化。这种设计确保了事务边界清晰，业务代码无需手动管理事务。