蓝图与项目结构

第一章：蓝图基础

什么是蓝图（Blueprint）？

蓝图是 Flask 中用于组织应用的一种方式。它允许你将应用分割成多个模块，每个模块有自己的路由、模板、静态文件等。蓝图本身不是一个应用，需要注册到应用上才能工作。

蓝图的优势

1. 模块化：将大型应用拆分成独立模块
2. 可复用：蓝图可以在多个应用中复用
3. 团队协作：不同团队可以负责不同蓝图
4. 代码组织：清晰的代码结构
5. 懒加载：按需加载模块

创建蓝图

from flask import Blueprint

# 创建蓝图
# 参数：蓝图名称, 导入名称（通常是 __name__）
main = Blueprint('main', __name__)

# 带配置的蓝图
auth = Blueprint(
    'auth',
    __name__,
    url_prefix='/auth',           # URL 前缀
    template_folder='templates',   # 模板目录
    static_folder='static',        # 静态文件目录
    static_url_path='/auth/static' # 静态文件 URL
)

# 定义路由
@main.route('/')
def index():
    return 'Hello from main blueprint!'

@auth.route('/login')
def login():
    return 'Login page'

# URL: /auth/login

注册蓝图

from flask import Flask
from blueprints.main import main
from blueprints.auth import auth
from blueprints.api import api

def create_app():
    app = Flask(__name__)

    # 注册蓝图
    app.register_blueprint(main)
    app.register_blueprint(auth)

    # 带 URL 前缀注册
    app.register_blueprint(api, url_prefix='/api/v1')

    # 带子域名注册
    # app.register_blueprint(admin, subdomain='admin')

    return app

蓝图资源

from flask import Blueprint, render_template, send_from_directory

# 蓝图可以有自己的模板和静态文件
admin = Blueprint(
    'admin',
    __name__,
    template_folder='templates',  # 相对于蓝图的模板目录
    static_folder='static',
    url_prefix='/admin'
)

@admin.route('/')
def dashboard():
    # 渲染蓝图的模板
    # 会先在蓝图的 templates 目录查找，再在应用的 templates 查找
    return render_template('admin/dashboard.html')

# 模板中引用蓝图的静态文件
# {{ url_for('admin.static', filename='css/admin.css') }}

第二章：蓝图进阶

蓝图钩子函数

from flask import Blueprint, g, request

api = Blueprint('api', __name__)

# 仅对本蓝图生效的钩子
@api.before_request
def before_request():
    """每个请求前执行"""
    g.start_time = time.time()

    # API 认证检查
    token = request.headers.get('Authorization')
    if not token:
        return jsonify({'error': 'Missing token'}), 401

@api.after_request
def after_request(response):
    """每个请求后执行"""
    duration = time.time() - g.start_time
    response.headers['X-Request-Duration'] = str(duration)
    return response

@api.teardown_request
def teardown_request(exception):
    """请求结束时执行"""
    if exception:
        app.logger.error(f'Request failed: {exception}')

# 应用级钩子（对所有蓝图生效）
@api.before_app_request
def before_app_request():
    """应用级：每个请求前执行"""
    pass

@api.after_app_request
def after_app_request(response):
    """应用级：每个请求后执行"""
    return response

蓝图错误处理

from flask import Blueprint, jsonify

api = Blueprint('api', __name__)

# 蓝图级错误处理
@api.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Resource not found'}), 404

@api.errorhandler(500)
def internal_error(error):
    return jsonify({'error': 'Internal server error'}), 500

# 自定义异常
class APIError(Exception):
    def __init__(self, message, status_code=400):
        self.message = message
        self.status_code = status_code

@api.errorhandler(APIError)
def handle_api_error(error):
    return jsonify({'error': error.message}), error.status_code

# 应用级错误处理
@api.app_errorhandler(404)
def app_not_found(error):
    """应用级 404 处理"""
    return jsonify({'error': 'Not found'}), 404

蓝图上下文处理器

from flask import Blueprint

admin = Blueprint('admin', __name__)

# 仅对本蓝图的模板生效
@admin.context_processor
def admin_context():
    return {
        'admin_name': 'Admin Panel',
        'admin_version': '2.0'
    }

# 应用级上下文处理器
@admin.app_context_processor
def app_context():
    return {
        'site_name': 'My Site'
    }

# 模板全局函数
@admin.app_template_global()
def format_datetime(value, format='%Y-%m-%d %H:%M'):
    return value.strftime(format)

# 模板过滤器
@admin.app_template_filter('currency')
def currency_filter(value):
    return f'¥{value:,.2f}'

URL 构建

from flask import url_for

# 在蓝图内构建 URL
@main.route('/about')
def about():
    # 引用本蓝图的路由
    home_url = url_for('.index')  # 或 url_for('main.index')

    # 引用其他蓝图的路由
    login_url = url_for('auth.login')
    api_url = url_for('api.get_users')

    return f'''
        Home: {home_url}
        Login: {login_url}
        API: {api_url}
    '''

# 在模板中使用
# <a href="{{ url_for('main.index') }}">Home</a>
# <a href="{{ url_for('auth.login') }}">Login</a>

第三章：项目结构模式

简单结构（小型项目）

myapp/
├── app.py              # 应用主文件
├── config.py           # 配置文件
├── requirements.txt    # 依赖
├── static/             # 静态文件
│   ├── css/
│   ├── js/
│   └── images/
├── templates/          # 模板文件
│   ├── base.html
│   ├── index.html
│   └── auth/
│       ├── login.html
│       └── register.html
└── models.py           # 数据模型

# app.py
from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy
from config import Config

app = Flask(__name__)
app.config.from_object(Config)

db = SQLAlchemy(app)

from models import User, Post  # 在 db 初始化后导入

@app.route('/')
def index():
    return render_template('index.html')

if __name__ == '__main__':
    app.run(debug=True)

功能型结构（中型项目）

myapp/
├── app/
│   ├── __init__.py          # 应用工厂
│   ├── models/              # 数据模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── post.py
│   ├── views/               # 视图函数（蓝图）
│   │   ├── __init__.py
│   │   ├── main.py
│   │   ├── auth.py
│   │   └── api.py
│   ├── forms/               # 表单
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   └── post.py
│   ├── services/            # 业务逻辑
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── email_service.py
│   ├── templates/           # 模板
│   │   ├── base.html
│   │   ├── main/
│   │   ├── auth/
│   │   └── errors/
│   ├── static/              # 静态文件
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   └── utils/               # 工具函数
│       ├── __init__.py
│       └── helpers.py
├── migrations/              # 数据库迁移
├── tests/                   # 测试
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_models.py
│   └── test_views.py
├── config.py                # 配置
├── requirements.txt         # 依赖
├── .env                     # 环境变量
└── run.py                   # 启动脚本

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from config import config

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    # 初始化扩展
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)

    # 注册蓝图
    from app.views.main import main
    from app.views.auth import auth
    from app.views.api import api

    app.register_blueprint(main)
    app.register_blueprint(auth, url_prefix='/auth')
    app.register_blueprint(api, url_prefix='/api')

    # 注册错误处理
    register_error_handlers(app)

    return app

def register_error_handlers(app):
    from flask import render_template

    @app.errorhandler(404)
    def not_found(error):
        return render_template('errors/404.html'), 404

    @app.errorhandler(500)
    def internal_error(error):
        return render_template('errors/500.html'), 500

# app/views/auth.py
from flask import Blueprint, render_template, redirect, url_for, flash
from flask_login import login_user, logout_user, login_required
from app.forms.auth import LoginForm, RegistrationForm
from app.models.user import User
from app import db

auth = Blueprint('auth', __name__)

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user and user.check_password(form.password.data):
            login_user(user, remember=form.remember.data)
            return redirect(url_for('main.index'))
        flash('Invalid email or password', 'error')
    return render_template('auth/login.html', form=form)

@auth.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('main.index'))

领域型结构（大型项目）

myapp/
├── app/
│   ├── __init__.py
│   ├── extensions.py        # 扩展初始化
│   ├── config.py            # 配置
│   │
│   ├── auth/                # 认证模块
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── views.py
│   │   ├── forms.py
│   │   ├── services.py
│   │   └── templates/
│   │       └── auth/
│   │
│   ├── blog/                # 博客模块
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── views.py
│   │   ├── forms.py
│   │   ├── services.py
│   │   └── templates/
│   │       └── blog/
│   │
│   ├── admin/               # 管理模块
│   │   ├── __init__.py
│   │   ├── views.py
│   │   └── templates/
│   │       └── admin/
│   │
│   ├── api/                 # API 模块
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   ├── posts.py
│   │   │   └── schemas.py
│   │   └── v2/
│   │
│   ├── common/              # 公共模块
│   │   ├── __init__.py
│   │   ├── decorators.py
│   │   ├── utils.py
│   │   └── exceptions.py
│   │
│   ├── templates/           # 公共模板
│   │   ├── base.html
│   │   └── errors/
│   │
│   └── static/              # 静态文件
│
├── migrations/
├── tests/
├── scripts/                 # 管理脚本
│   ├── create_admin.py
│   └── seed_data.py
├── docker/                  # Docker 配置
│   ├── Dockerfile
│   └── docker-compose.yml
├── docs/                    # 文档
├── requirements/            # 分环境依赖
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env.example
├── .gitignore
├── Makefile
└── run.py

# app/__init__.py
from flask import Flask
from app.extensions import db, migrate, login_manager, mail, cache
from app.config import config

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    # 初始化扩展
    register_extensions(app)

    # 注册蓝图
    register_blueprints(app)

    # 注册错误处理
    register_error_handlers(app)

    # 注册 CLI 命令
    register_commands(app)

    # 注册 Shell 上下文
    register_shell_context(app)

    return app

def register_extensions(app):
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    mail.init_app(app)
    cache.init_app(app)

def register_blueprints(app):
    from app.auth import auth_bp
    from app.blog import blog_bp
    from app.admin import admin_bp
    from app.api.v1 import api_v1_bp

    app.register_blueprint(auth_bp)
    app.register_blueprint(blog_bp)
    app.register_blueprint(admin_bp, url_prefix='/admin')
    app.register_blueprint(api_v1_bp, url_prefix='/api/v1')

def register_error_handlers(app):
    from app.common.exceptions import register_handlers
    register_handlers(app)

def register_commands(app):
    from app.commands import init_db, create_admin
    app.cli.add_command(init_db)
    app.cli.add_command(create_admin)

def register_shell_context(app):
    from app.auth.models import User
    from app.blog.models import Post, Category

    @app.shell_context_processor
    def make_shell_context():
        return {
            'db': db,
            'User': User,
            'Post': Post,
            'Category': Category
        }

# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from flask_caching import Cache
from flask_wtf.csrf import CSRFProtect

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
mail = Mail()
cache = Cache()
csrf = CSRFProtect()

# 配置 LoginManager
login_manager.login_view = 'auth.login'
login_manager.login_message = '请先登录'
login_manager.login_message_category = 'warning'

@login_manager.user_loader
def load_user(user_id):
    from app.auth.models import User
    return User.query.get(int(user_id))

# app/auth/__init__.py
from flask import Blueprint

auth_bp = Blueprint('auth', __name__, template_folder='templates')

from app.auth import views  # 导入视图

第四章：应用工厂模式

为什么使用应用工厂

1. 测试：可以创建不同配置的应用实例
2. 多实例：可以运行多个不同配置的应用
3. 延迟配置：配置在运行时决定
4. 循环导入：避免循环导入问题

基本工厂函数

# app/__init__.py
from flask import Flask

def create_app(config_name=None):
    """应用工厂函数"""
    app = Flask(__name__)

    # 加载配置
    if config_name is None:
        config_name = os.environ.get('FLASK_CONFIG', 'default')

    from config import config
    app.config.from_object(config[config_name])

    # 初始化扩展
    configure_extensions(app)

    # 注册蓝图
    configure_blueprints(app)

    # 配置日志
    configure_logging(app)

    # 注册错误处理
    configure_error_handlers(app)

    return app

def configure_extensions(app):
    """配置扩展"""
    from app.extensions import db, migrate, login_manager

    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)

    # 配置登录管理器
    login_manager.login_view = 'auth.login'

    @login_manager.user_loader
    def load_user(user_id):
        from app.models import User
        return User.query.get(int(user_id))

def configure_blueprints(app):
    """注册蓝图"""
    from app.views.main import main
    from app.views.auth import auth

    app.register_blueprint(main)
    app.register_blueprint(auth, url_prefix='/auth')

def configure_logging(app):
    """配置日志"""
    import logging
    from logging.handlers import RotatingFileHandler

    if not app.debug:
        file_handler = RotatingFileHandler(
            'logs/app.log',
            maxBytes=10240,
            backupCount=10
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('Application startup')

def configure_error_handlers(app):
    """配置错误处理"""
    from flask import render_template

    @app.errorhandler(404)
    def not_found(error):
        return render_template('errors/404.html'), 404

    @app.errorhandler(500)
    def internal_error(error):
        from app.extensions import db
        db.session.rollback()
        return render_template('errors/500.html'), 500

配置类

# config.py
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    """基础配置"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    # 邮件配置
    MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.example.com')
    MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')

    # 缓存配置
    CACHE_TYPE = 'simple'

    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config):
    """开发环境配置"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'dev.db')

    # 开发环境特定配置
    TEMPLATES_AUTO_RELOAD = True
    SQLALCHEMY_ECHO = True

class TestingConfig(Config):
    """测试环境配置"""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
    WTF_CSRF_ENABLED = False

class ProductionConfig(Config):
    """生产环境配置"""
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

    # 生产环境特定配置
    SSL_REDIRECT = True
    SESSION_COOKIE_SECURE = True
    REMEMBER_COOKIE_SECURE = True

    @classmethod
    def init_app(cls, app):
        Config.init_app(app)

        # 生产环境日志配置
        import logging
        from logging.handlers import SMTPHandler

        credentials = None
        secure = None
        if getattr(cls, 'MAIL_USERNAME', None) is not None:
            credentials = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
            if getattr(cls, 'MAIL_USE_TLS', None):
                secure = ()

        mail_handler = SMTPHandler(
            mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
            fromaddr=cls.MAIL_USERNAME,
            toaddrs=[os.environ.get('ADMIN_EMAIL')],
            subject='Application Error',
            credentials=credentials,
            secure=secure
        )
        mail_handler.setLevel(logging.ERROR)
        app.logger.addHandler(mail_handler)

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

启动脚本

# run.py
import os
from app import create_app, db
from app.models import User, Post

app = create_app(os.environ.get('FLASK_CONFIG') or 'default')

@app.shell_context_processor
def make_shell_context():
    """Shell 上下文"""
    return dict(db=db, User=User, Post=Post)

@app.cli.command()
def deploy():
    """部署命令"""
    from flask_migrate import upgrade
    upgrade()

if __name__ == '__main__':
    app.run()

第五章：CLI 命令

自定义 CLI 命令

# app/commands.py
import click
from flask.cli import with_appcontext
from app.extensions import db

@click.command('init-db')
@with_appcontext
def init_db():
    """初始化数据库"""
    db.create_all()
    click.echo('Database initialized.')

@click.command('create-admin')
@click.argument('username')
@click.argument('email')
@click.password_option()
@with_appcontext
def create_admin(username, email, password):
    """创建管理员用户"""
    from app.models import User

    user = User(username=username, email=email, is_admin=True)
    user.set_password(password)
    db.session.add(user)
    db.session.commit()

    click.echo(f'Admin user {username} created.')

@click.command('seed-data')
@with_appcontext
def seed_data():
    """填充测试数据"""
    from app.models import User, Post, Category
    from faker import Faker

    fake = Faker('zh_CN')

    # 创建分类
    categories = []
    for name in ['技术', '生活', '随笔']:
        cat = Category(name=name)
        db.session.add(cat)
        categories.append(cat)

    # 创建用户和文章
    for i in range(10):
        user = User(
            username=fake.user_name(),
            email=fake.email()
        )
        user.set_password('password')
        db.session.add(user)

        for j in range(5):
            post = Post(
                title=fake.sentence(),
                content=fake.text(max_nb_chars=1000),
                author=user,
                category=fake.random_element(categories)
            )
            db.session.add(post)

    db.session.commit()
    click.echo('Test data seeded.')

# 命令组
@click.group()
def translate():
    """翻译和本地化命令"""
    pass

@translate.command()
def update():
    """更新翻译文件"""
    click.echo('Updating translations...')

@translate.command()
def compile():
    """编译翻译文件"""
    click.echo('Compiling translations...')

# app/__init__.py
def register_commands(app):
    from app.commands import init_db, create_admin, seed_data, translate

    app.cli.add_command(init_db)
    app.cli.add_command(create_admin)
    app.cli.add_command(seed_data)
    app.cli.add_command(translate)

使用命令

# 查看可用命令
flask --help

# 初始化数据库
flask init-db

# 创建管理员
flask create-admin admin admin@example.com

# 填充测试数据
flask seed-data

# 翻译命令组
flask translate update
flask translate compile

第六章：大型项目最佳实践

配置管理

# app/config.py
import os
from pathlib import Path

class BaseConfig:
    """基础配置"""
    BASE_DIR = Path(__file__).resolve().parent.parent

    # 安全配置
    SECRET_KEY = os.environ.get('SECRET_KEY')
    if not SECRET_KEY:
        raise ValueError('SECRET_KEY environment variable is not set')

    # 数据库
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    # 会话
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = 'Lax'

    # 上传
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB
    UPLOAD_FOLDER = BASE_DIR / 'uploads'

    # 分页
    POSTS_PER_PAGE = 20

    @classmethod
    def init_app(cls, app):
        # 确保上传目录存在
        cls.UPLOAD_FOLDER.mkdir(exist_ok=True)

class DevelopmentConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'
    SESSION_COOKIE_SECURE = False

class ProductionConfig(BaseConfig):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

    # 使用 Redis 存储会话
    SESSION_TYPE = 'redis'
    SESSION_REDIS = os.environ.get('REDIS_URL')

    @classmethod
    def init_app(cls, app):
        super().init_app(app)

        # 配置代理
        from werkzeug.middleware.proxy_fix import ProxyFix
        app.wsgi_app = ProxyFix(
            app.wsgi_app,
            x_for=1,
            x_proto=1,
            x_host=1,
            x_prefix=1
        )

依赖注入

# app/services/__init__.py
class ServiceContainer:
    """服务容器"""
    _services = {}

    @classmethod
    def register(cls, name, service):
        cls._services[name] = service

    @classmethod
    def get(cls, name):
        return cls._services.get(name)

# app/services/user_service.py
class UserService:
    def __init__(self, db, mail):
        self.db = db
        self.mail = mail

    def create_user(self, username, email, password):
        from app.models import User
        user = User(username=username, email=email)
        user.set_password(password)
        self.db.session.add(user)
        self.db.session.commit()
        return user

    def send_welcome_email(self, user):
        from flask_mail import Message
        msg = Message(
            '欢迎加入',
            recipients=[user.email]
        )
        msg.body = f'你好 {user.username}，欢迎加入我们！'
        self.mail.send(msg)

# 在应用初始化时注册服务
def configure_services(app):
    from app.extensions import db, mail
    from app.services.user_service import UserService

    with app.app_context():
        ServiceContainer.register('user', UserService(db, mail))

信号使用

# app/signals.py
from blinker import Namespace

# 创建信号命名空间
signals = Namespace()

# 定义信号
user_registered = signals.signal('user-registered')
user_logged_in = signals.signal('user-logged-in')
post_created = signals.signal('post-created')
post_deleted = signals.signal('post-deleted')

# app/auth/views.py
from app.signals import user_registered, user_logged_in

@auth.route('/register', methods=['POST'])
def register():
    # ... 创建用户 ...

    # 发送信号
    user_registered.send(current_app._get_current_object(), user=user)

    return redirect(url_for('main.index'))

# app/listeners.py
from app.signals import user_registered, user_logged_in

@user_registered.connect
def on_user_registered(sender, user):
    """用户注册后发送欢迎邮件"""
    from app.services import ServiceContainer
    user_service = ServiceContainer.get('user')
    user_service.send_welcome_email(user)

@user_logged_in.connect
def on_user_logged_in(sender, user):
    """用户登录后更新登录时间"""
    from datetime import datetime
    from app.extensions import db

    user.last_login = datetime.utcnow()
    db.session.commit()

中间件

# app/middleware.py
from flask import request, g
import time
import uuid

class RequestIdMiddleware:
    """请求 ID 中间件"""
    def __init__(self, app):
        self.app = app
        app.before_request(self.before_request)
        app.after_request(self.after_request)

    def before_request(self):
        g.request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())
        g.start_time = time.time()

    def after_request(self, response):
        response.headers['X-Request-ID'] = g.request_id
        duration = time.time() - g.start_time
        response.headers['X-Request-Duration'] = f'{duration:.4f}s'
        return response

class MaintenanceModeMiddleware:
    """维护模式中间件"""
    def __init__(self, app):
        self.app = app
        app.before_request(self.check_maintenance)

    def check_maintenance(self):
        from flask import render_template, current_app

        if current_app.config.get('MAINTENANCE_MODE'):
            return render_template('maintenance.html'), 503

# 在应用初始化时使用
def configure_middleware(app):
    RequestIdMiddleware(app)
    MaintenanceModeMiddleware(app)

任务队列

# app/tasks/__init__.py
from celery import Celery

celery = Celery('app')

def make_celery(app):
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

# app/tasks/email.py
from app.tasks import celery
from flask_mail import Message
from app.extensions import mail

@celery.task
def send_async_email(subject, recipients, body, html=None):
    """异步发送邮件"""
    msg = Message(subject, recipients=recipients)
    msg.body = body
    if html:
        msg.html = html
    mail.send(msg)

@celery.task
def send_welcome_email(user_id):
    """发送欢迎邮件"""
    from app.models import User
    user = User.query.get(user_id)
    if user:
        send_async_email.delay(
            '欢迎加入',
            [user.email],
            f'你好 {user.username}，欢迎加入！'
        )

# 使用
# send_welcome_email.delay(user.id)

总结

本章详细介绍了 Flask 蓝图与项目结构：

1. 蓝图基础：创建、注册、资源管理
2. 蓝图进阶：钩子、错误处理、上下文处理器
3. 项目结构：简单、功能型、领域型结构
4. 应用工厂：工厂函数、配置管理
5. CLI 命令：自定义命令、命令组
6. 最佳实践：配置管理、依赖注入、信号、中间件、任务队列

下一章我们将学习 RESTful API 开发。