【结构】 - src/core/:database/response/jwt/password/security/redis_client/mail/scheduler - src/common/:base_model/base_db/forms(base_form+fields)/constants/utils - src/models_registry.py:模型聚合器(Flask-Migrate 全量表识别) 【绞杀者 shim】extensions/、utils/、application/constants|forms|models 打兼容转发 【兼容修复】应用工厂抽取至 application/factory.py(P2 前置,P1 已完成) 【基线修复】application/models(25) + middleware(7) 原被嵌套 .gitignore(*) 误排除出版本库,已补入(base_model/base_db 现为 shim,原版在 src/common/) 【登记既有缺陷】 1. flask db CLI 因 .env FLASK_VERSION 与 --version 冲突报错 → Makefile 已加 FLASK_SKIP_DOTENV=1 2. 本机 Redis 未设密码但 REDIS_AUTH=True,/captcha 报 AUTH 错误(环境问题) 【验收】22 模型表全量识别;src/ flake8 零告警;pytest 5 passed;应用启动; /login 表单校验、/user/page 鉴权拦截冒烟正常 Co-Authored-By: Claude <noreply@anthropic.com>
242 lines
7.3 KiB
Python
242 lines
7.3 KiB
Python
"""
|
||
AOP 事务拦截器(企业级实现)
|
||
|
||
设计理念:
|
||
1. 零代码侵入:业务代码无需任何事务管理代码
|
||
2. 自动读写分离:GET 不提交,POST/PUT/DELETE 自动提交
|
||
3. 统一异常处理:所有异常自动回滚
|
||
4. 性能监控:记录事务执行时间
|
||
|
||
使用方式:
|
||
在 Flask 应用初始化时调用 register_transaction(app) 即可。
|
||
业务代码无需任何改动。
|
||
|
||
Example:
|
||
# app/__init__.py
|
||
from application.middleware.transaction import register_transaction
|
||
|
||
def create_app():
|
||
app = Flask(__name__)
|
||
register_transaction(app) # 一行代码搞定所有事务
|
||
return app
|
||
"""
|
||
|
||
from flask import request, g
|
||
from extensions import db
|
||
import logging
|
||
import time
|
||
from functools import wraps
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ============================================================
|
||
# 核心:AOP 事务拦截器
|
||
# ============================================================
|
||
|
||
def register_transaction(app):
|
||
"""
|
||
注册 AOP 事务拦截器到 Flask 应用
|
||
|
||
在请求生命周期中自动管理事务:
|
||
- before_request: 开启事务标记
|
||
- after_request: 自动提交(POST/PUT/DELETE)
|
||
- teardown_request: 异常回滚 + 资源清理
|
||
|
||
Args:
|
||
app: Flask 应用实例
|
||
|
||
Example:
|
||
register_transaction(app)
|
||
"""
|
||
|
||
@app.before_request
|
||
def begin_transaction():
|
||
"""请求开始:记录开始时间,开启事务标记"""
|
||
g.request_start_time = time.time()
|
||
g.transaction_active = True
|
||
|
||
logger.debug(
|
||
f"请求开始, method: {request.method}, path: {request.path}, "
|
||
f"ip: {request.remote_addr}"
|
||
)
|
||
|
||
@app.after_request
|
||
def commit_transaction(response):
|
||
"""
|
||
请求结束:自动提交事务
|
||
|
||
提交规则:
|
||
1. 只对写操作(POST/PUT/DELETE/PATCH)提交
|
||
2. 只在响应成功(2xx)且业务成功(code=0)时提交
|
||
3. GET/OPTIONS/HEAD 不提交(只读)
|
||
4. 业务失败(R.failed)不提交,自动回滚
|
||
"""
|
||
# 只对写操作处理事务
|
||
if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
|
||
should_commit = 200 <= response.status_code < 300
|
||
|
||
# 检查业务状态码:R.failed() 返回 HTTP 200 但 code=1/ok=False
|
||
if should_commit:
|
||
try:
|
||
data = response.get_json(silent=True)
|
||
if data and data.get('code', 0) != 0:
|
||
should_commit = False
|
||
except Exception:
|
||
pass
|
||
|
||
if should_commit:
|
||
try:
|
||
if db.session.is_active:
|
||
db.session.commit()
|
||
logger.debug(
|
||
f"事务提交成功, method: {request.method}, "
|
||
f"path: {request.path}, status: {response.status_code}"
|
||
)
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
logger.error(
|
||
f"事务提交失败, method: {request.method}, "
|
||
f"path: {request.path}, error: {str(e)}",
|
||
exc_info=True
|
||
)
|
||
from utils import R
|
||
return R.failed(msg="数据处理失败"), 500
|
||
else:
|
||
# 业务失败或非2xx,回滚未提交的变更
|
||
try:
|
||
if db.session.is_active:
|
||
db.session.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
# 记录响应日志
|
||
elapsed = time.time() - g.get('request_start_time', time.time())
|
||
logger.debug(
|
||
f"请求结束, method: {request.method}, path: {request.path}, "
|
||
f"status: {response.status_code}, elapsed: {elapsed:.3f}s"
|
||
)
|
||
|
||
return response
|
||
|
||
@app.teardown_request
|
||
def cleanup_transaction(exception=None):
|
||
"""
|
||
请求结束:清理资源
|
||
|
||
发生异常时自动回滚事务,确保数据一致性。
|
||
"""
|
||
if exception is not None:
|
||
try:
|
||
db.session.rollback()
|
||
logger.error(
|
||
f"请求异常,事务已回滚, method: {request.method}, "
|
||
f"path: {request.path}, error: {str(exception)}",
|
||
exc_info=True
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"事务回滚失败: {str(e)}", exc_info=True)
|
||
|
||
# 清理请求标记
|
||
g.pop('transaction_active', None)
|
||
g.pop('request_start_time', None)
|
||
|
||
logger.info("AOP 事务拦截器注册成功")
|
||
|
||
|
||
# ============================================================
|
||
# 备用:精细控制装饰器(特殊场景使用)
|
||
# ============================================================
|
||
|
||
def transactional(func):
|
||
"""
|
||
事务装饰器(备用)
|
||
|
||
适用于需要精细控制事务边界的特殊场景。
|
||
大多数情况下,使用 AOP 拦截器即可。
|
||
|
||
Example:
|
||
@transactional
|
||
def complex_business():
|
||
# 精细控制的事务逻辑
|
||
pass
|
||
"""
|
||
|
||
@wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
if g.get('disable_auto_commit', False):
|
||
return func(*args, **kwargs)
|
||
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
if not g.get('transaction_active', False):
|
||
db.session.commit()
|
||
return result
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True)
|
||
raise
|
||
|
||
return wrapper
|
||
|
||
|
||
def transactional_response(func):
|
||
"""
|
||
事务装饰器(带返回 R 对象)
|
||
|
||
用于需要直接返回 R 对象的接口。
|
||
|
||
Example:
|
||
@transactional_response
|
||
def add_position():
|
||
position.save()
|
||
return R.ok(msg="添加成功")
|
||
"""
|
||
|
||
@wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
from utils import R
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
if not g.get('transaction_active', False):
|
||
db.session.commit()
|
||
return result
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True)
|
||
return R.failed(msg=f"操作失败: {str(e)}")
|
||
|
||
return wrapper
|
||
|
||
|
||
# ============================================================
|
||
# 备用:事务上下文管理器(特殊场景使用)
|
||
# ============================================================
|
||
|
||
class TransactionContext:
|
||
"""
|
||
事务上下文管理器
|
||
|
||
用于需要精细控制事务的特殊场景。
|
||
|
||
Example:
|
||
with TransactionContext():
|
||
position.save()
|
||
user.save()
|
||
# 自动提交或回滚
|
||
"""
|
||
|
||
def __enter__(self):
|
||
g.disable_auto_commit = True
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
g.disable_auto_commit = False
|
||
if exc_type is None:
|
||
if not g.get('transaction_active', False):
|
||
db.session.commit()
|
||
else:
|
||
db.session.rollback()
|
||
logger.error(f"事务上下文回滚: {str(exc_val)}", exc_info=True)
|
||
return False
|