Become a sponsor
应用工厂与扩展初始化
本章概要
Flask 应用工厂模式（create_app()）是项目的核心初始化机制。本文详解工厂函数的初始化顺序、扩展注册方式、依赖注入机制，帮助开发者理解应用启动全流程。
应用工厂模式
项目采用 Flask 标准的应用工厂模式，通过 create_app() 函数创建和配置应用实例：
python
# src/app.py
from flask import Flask
from flask_cors import CORS
from werkzeug.middleware.proxy_fix import ProxyFix
from src.middleware import register_middleware, register_transaction, register_error_handlers, register_operation_log
from src.routes.v1 import register_v1
from src.extensions import register_extends
from config import DatabaseConfig, REDIS_URL
from src.common.utils.json import JSONEncoder
def create_app():
"""
创建并配置 Flask 应用实例
配置流程：
1. 创建 Flask 核心对象
2. 配置 Redis 连接
3. 加载数据库配置
4. 设置 JSON 序列化器
5. 配置跨域资源共享
6. 注册操作日志与事务拦截器
7. 注册 API v1 路由
8. 初始化第三方扩展
9. 注册中间件
10. 注入依赖
Returns:
Flask: 配置完成的应用实例
"""
# 1. 创建 Flask 核心对象
app = Flask(__name__, template_folder='templates', static_folder='../static')
# 2. 配置 Redis
app.config['REDIS_URL'] = REDIS_URL
# 3. 加载数据库配置
app.config.from_object(DatabaseConfig)
# 4. 自定义 JSON 序列化器
app.json = JSONEncoder(app)
# 5. 配置 CORS
CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=False)
# 6. 注册操作日志 + AOP 事务拦截器
register_operation_log(app)
register_transaction(app)
# 7. 注册 API v1 路由
register_v1(app)
# 8. 初始化扩展（数据库、Redis、调度器等）
register_extends(app)
# 9. 注册中间件（登录校验、演示模式、异常处理）
register_middleware(app)
# 10. 注入依赖
_inject_dependencies(app)
return app初始化顺序的重要性
初始化顺序经过精心设计，各步骤之间存在依赖关系：
Flask 核心对象
↓
配置加载（数据库、Redis）
↓
JSON 序列化器
↓
CORS 跨域
↓
操作日志 + 事务拦截器（须在路由之前）
↓
路由注册（Blueprint 挂载）
↓
扩展初始化（db.init_app、redis 连接）
↓
中间件注册（登录校验、演示模式）
↓
依赖注入（权限提供方、操作日志写入器）关键顺序
register_operation_log必须在register_transaction之前，确保操作日志在事务提交前记录register_v1（路由注册）必须在register_middleware（登录校验）之前，否则中间件无法拦截已注册的路由register_extends（扩展初始化）必须在路由处理请求之前完成
扩展注册（init_app 模式）
Flask 扩展采用 init_app 模式，将扩展实例与应用绑定：
python
# src/extensions.py
from src.core.database import db
from src.core.redis_client import redis
from src.core.scheduler import scheduler
def register_extends(app):
"""初始化第三方扩展"""
# Flask-SQLAlchemy
db.init_app(app)
# Redis 连接
redis.init_app(app)
# APScheduler 定时任务
scheduler.init_app(app)python
# src/core/database.py
from flask_sqlalchemy import SQLAlchemy
# 创建扩展实例（不绑定应用）
db = SQLAlchemy()这种模式的优势：
- 延迟初始化：扩展实例在模块加载时创建，但绑定应用发生在
create_app()中 - 测试友好：测试时可以创建不同的应用实例，扩展自动绑定到测试应用
- 避免循环导入：扩展实例定义在
src/core/，应用工厂在src/app.py，解耦清晰
路由注册（Blueprint 模式）
所有业务蓝图通过 src/routes/v1.py 统一注册到 /api/v1 前缀：
python
# src/routes/v1.py
from flask import Blueprint
from src.modules.system.position.router import position_blue
from src.modules.system.user.router import user_blue
from src.modules.system.role.router import role_blue
# ... 其他模块导入
# 全部业务蓝图（按领域分组排序）
BLUEPRINTS = [
# 系统管理模块
user_blue, role_blue, menu_blue, dept_blue, position_blue, level_blue,
# 内容管理模块
article_blue, category_blue, link_blue,
# 数据管理模块
dict_blue, config_blue, notice_blue, param_blue, city_blue,
# 监控管理模块
job_blue, job_log_blue, login_log_blue, operation_log_blue,
# 认证模块
auth_blue,
# 文件上传
upload_blue,
]
def register_v1(app):
"""
注册 API v1 路由（所有业务蓝图挂载到 /api/v1）
Args:
app: Flask 应用实例
"""
v1 = Blueprint('v1', __name__, url_prefix='/api/v1')
for blue in BLUEPRINTS:
v1.register_blueprint(blue)
app.register_blueprint(v1)中间件注册
中间件通过 before_request / after_request / teardown_request 钩子实现：
python
# src/middleware/__init__.py
from src.middleware.auth import check_login, is_whitelisted, register_middleware
from src.middleware.demo import check_demo
from src.middleware.transaction import register_transaction, transactional, TransactionContext
from src.middleware.operation_log import register_operation_log, operation_log
from src.middleware.errors import register_error_handlers各中间件在 src/app.py 的 create_app() 中按顺序注册：
python
# src/app.py - create_app()
register_operation_log(app) # 操作日志记录
register_transaction(app) # AOP 事务管理
register_v1(app) # 路由注册
register_extends(app) # 扩展初始化
register_middleware(app) # 登录校验 + 白名单
register_error_handlers(app) # 全局异常处理中间件执行顺序（请求进入时）：
请求进入
↓
login_required（检查 JWT Token）
↓
check_permission（RBAC 权限校验，装饰器层面）
↓
check_demo（演示模式拦截，装饰器层面）
↓
路由函数（Router）
↓
Service → Repository → 数据库
↓
返回响应依赖注入
_inject_dependencies 将权限查询函数注入到核心模块，解耦业务模块与核心模块：
python
# src/app.py
def _inject_dependencies(app):
"""
注入依赖（权限提供方 / 操作日志写入器）
通过 app.extensions 字典传递依赖，避免核心模块直接导入业务模块。
"""
from src.modules.system.menu.service import get_user_permissions
from src.modules.operation_log.service import save_operation_log
app.extensions['permission_provider'] = get_user_permissions
app.extensions['operation_log_writer'] = save_operation_logpython
# src/core/permission.py
from flask import current_app
def check_permission(permission: str):
"""权限校验装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 通过 current_app 获取注入的权限查询函数
permission_provider = current_app.extensions.get('permission_provider')
if permission_provider:
user_perms = permission_provider(user_id)
if permission not in user_perms:
return R.failed(msg="权限不足", code=403)
return func(*args, **kwargs)
return wrapper
return decorator入口文件
python
# app.py（项目根目录）
from src.app import create_app
app = create_app()
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8021, debug=True)生产环境使用 Gunicorn 启动：
bash
# 4 个工作进程，每个进程 4 个线程
gunicorn -w 4 -t 4 "app:app" --bind 0.0.0.0:8021总结
项目采用 Flask 标准的应用工厂模式，通过 create_app() 按序完成配置加载、扩展初始化、路由注册、中间件注册、依赖注入。扩展采用 init_app 模式延迟绑定，路由通过 Blueprint 模块化组织，中间件通过 before_request / after_request 钩子链式处理。这种设计确保了模块解耦、测试友好、初始化顺序可控。