Flask蓝图(Blueprint)

蓝图是Flask中用于组织大型应用的模块化方法,允许将应用分解为可重用的组件。

1. 蓝图基础概念

1.1 为什么需要蓝图?

当Flask应用变得复杂时,将所有代码放在一个文件中会导致:

  • 代码难以维护
  • 路由冲突
  • 功能模块耦合
  • 团队协作困难

蓝图解决了这些问题,它允许将应用分解为多个模块,每个模块可以独立开发。

1.2 蓝图结构对比

# ❌ 不使用蓝图的应用结构(单个文件)
myapp/
    app.py  # 所有代码都在这里,上千行代码

# ✅ 使用蓝图的应用结构(模块化)
myapp/
    app.py              # 应用工厂和配置
    __init__.py         # 包初始化
    auth/               # 认证模块
        __init__.py
        routes.py
        forms.py
        models.py
        templates/
            auth/
                login.html
                register.html
    blog/               # 博客模块
        __init__.py
        routes.py
        models.py
        templates/
            blog/
                index.html
                post.html
    admin/              # 管理后台模块
        __init__.py
        routes.py
    static/             # 静态文件
        css/
        js/
        images/
    templates/          # 通用模板
        base.html
        layout.html

2. 创建第一个蓝图

2.1 基础蓝图定义

# auth/__init__.py
from flask import Blueprint

# 创建蓝图对象
auth_bp = Blueprint(
    'auth',  # 蓝图名称,用于反向解析URL
    __name__,  # 蓝图的模块名
    url_prefix='/auth',  # URL前缀,所有路由都以/auth开头
    template_folder='templates',  # 蓝图的模板文件夹
    static_folder='static',  # 蓝图的静态文件文件夹
    static_url_path='/auth/static'  # 静态文件URL前缀
)

# 导入路由(避免循环导入)
from . import routes

2.2 定义蓝图路由

# auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from . import auth_bp

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    """登录页面"""
    if request.method == 'POST':
        # 处理登录逻辑
        flash('登录成功!', 'success')
        return redirect(url_for('main.index'))

    return render_template('auth/login.html')

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    """注册页面"""
    if request.method == 'POST':
        # 处理注册逻辑
        flash('注册成功!', 'success')
        return redirect(url_for('auth.login'))

    return render_template('auth/register.html')

@auth_bp.route('/logout')
def logout():
    """退出登录"""
    flash('您已退出登录', 'info')
    return redirect(url_for('auth.login'))

@auth_bp.route('/profile')
def profile():
    """用户个人资料"""
    return render_template('auth/profile.html')

@auth_bp.route('/forgot-password', methods=['GET', 'POST'])
def forgot_password():
    """忘记密码"""
    return render_template('auth/forgot_password.html')

# 错误处理器(蓝图级别)
@auth_bp.app_errorhandler(404)
def auth_not_found(error):
    """认证模块的404错误处理器"""
    return render_template('auth/404.html'), 404

3. 博客蓝图示例

# blog/__init__.py
from flask import Blueprint

blog_bp = Blueprint(
    'blog',
    __name__,
    url_prefix='/blog',
    template_folder='templates',
    static_folder='static',
    static_url_path='/blog/static'
)

# 导入路由和模型
from . import routes, models
# blog/routes.py
from flask import render_template, request, redirect, url_for, flash, abort
from . import blog_bp
from .models import Post, Category, Comment

@blog_bp.route('/')
def index():
    """博客首页"""
    page = request.args.get('page', 1, type=int)
    per_page = 10

    posts = Post.query.filter_by(is_published=True).order_by(
        Post.created_at.desc()
    ).paginate(page=page, per_page=per_page)

    categories = Category.query.all()

    return render_template('blog/index.html',
                         posts=posts,
                         categories=categories)

@blog_bp.route('/post/<slug>')
def post_detail(slug):
    """文章详情页"""
    post = Post.query.filter_by(slug=slug, is_published=True).first_or_404()

    # 增加浏览量
    post.views += 1
    from .. import db
    db.session.commit()

    return render_template('blog/post.html', post=post)

@blog_bp.route('/category/<slug>')
def category_posts(slug):
    """分类文章列表"""
    category = Category.query.filter_by(slug=slug).first_or_404()

    page = request.args.get('page', 1, type=int)
    per_page = 10

    posts = Post.query.filter_by(
        is_published=True
    ).filter(
        Post.categories.any(slug=slug)
    ).order_by(
        Post.created_at.desc()
    ).paginate(page=page, per_page=per_page)

    return render_template('blog/category.html',
                         category=category,
                         posts=posts)

@blog_bp.route('/search')
def search():
    """搜索文章"""
    query = request.args.get('q', '')
    page = request.args.get('page', 1, type=int)

    if not query:
        return redirect(url_for('blog.index'))

    posts = Post.query.filter(
        Post.is_published == True,
        Post.title.contains(query) | Post.content.contains(query)
    ).order_by(Post.created_at.desc()).paginate(page=page)

    return render_template('blog/search.html',
                         query=query,
                         posts=posts)

@blog_bp.route('/archive/<int:year>/<int:month>')
def archive(year, month):
    """文章归档"""
    from datetime import datetime
    from sqlalchemy import extract

    page = request.args.get('page', 1, type=int)

    posts = Post.query.filter(
        Post.is_published == True,
        extract('year', Post.created_at) == year,
        extract('month', Post.created_at) == month
    ).order_by(Post.created_at.desc()).paginate(page=page)

    archive_title = f"{year}年{month}月"

    return render_template('blog/archive.html',
                         posts=posts,
                         archive_title=archive_title,
                         year=year,
                         month=month)

4. 管理后台蓝图

# admin/__init__.py
from flask import Blueprint

admin_bp = Blueprint(
    'admin',
    __name__,
    url_prefix='/admin',
    template_folder='templates',
    static_folder='static'
)
# admin/routes.py
from flask import render_template, redirect, url_for, request, flash, abort
from flask_login import login_required, current_user
from . import admin_bp

@admin_bp.before_request
def check_admin():
    """管理后台的请求前检查"""
    from flask_login import current_user

    if not current_user.is_authenticated:
        return redirect(url_for('auth.login'))

    if not current_user.is_admin:
        flash('您没有访问权限', 'error')
        return redirect(url_for('main.index'))

@admin_bp.route('/')
@login_required
def dashboard():
    """管理后台首页"""
    return render_template('admin/dashboard.html')

@admin_bp.route('/users')
@login_required
def users():
    """用户管理"""
    from ..models import User
    page = request.args.get('page', 1, type=int)

    users = User.query.order_by(User.created_at.desc()).paginate(page=page)

    return render_template('admin/users.html', users=users)

@admin_bp.route('/posts')
@login_required
def posts():
    """文章管理"""
    from ..models import Post
    page = request.args.get('page', 1, type=int)
    status = request.args.get('status', 'all')

    query = Post.query

    if status == 'published':
        query = query.filter_by(is_published=True)
    elif status == 'draft':
        query = query.filter_by(is_published=False)

    posts = query.order_by(Post.created_at.desc()).paginate(page=page)

    return render_template('admin/posts.html', posts=posts, status=status)

@admin_bp.route('/comments')
@login_required
def comments():
    """评论管理"""
    from ..models import Comment
    page = request.args.get('page', 1, type=int)
    status = request.args.get('status', 'all')

    query = Comment.query

    if status == 'pending':
        query = query.filter_by(is_approved=False)
    elif status == 'approved':
        query = query.filter_by(is_approved=True)

    comments = query.order_by(Comment.created_at.desc()).paginate(page=page)

    return render_template('admin/comments.html',
                         comments=comments,
                         status=status)

@admin_bp.route('/settings', methods=['GET', 'POST'])
@login_required
def settings():
    """系统设置"""
    if request.method == 'POST':
        # 处理设置更新
        flash('设置已保存', 'success')
        return redirect(url_for('admin.settings'))

    return render_template('admin/settings.html')

5. 应用工厂模式

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

# 创建扩展实例(但不初始化)
db = SQLAlchemy()
login_manager = LoginManager()
migrate = Migrate()

def create_app(config_class=Config):
    """应用工厂函数"""
    app = Flask(__name__)
    app.config.from_object(config_class)

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

    # 配置登录管理器
    login_manager.login_view = 'auth.login'
    login_manager.login_message = '请先登录'
    login_manager.login_message_category = 'info'

    # 注册蓝图
    from app.auth import auth_bp
    from app.blog import blog_bp
    from app.admin import admin_bp
    from app.main import main_bp

    app.register_blueprint(auth_bp)
    app.register_blueprint(blog_bp)
    app.register_blueprint(admin_bp)
    app.register_blueprint(main_bp)

    # 注册上下文处理器
    @app.context_processor
    def inject_global_vars():
        """注入全局变量到所有模板"""
        from app.models import Category
        return dict(
            site_name=app.config.get('SITE_NAME', '我的网站'),
            categories=Category.query.all()
        )

    # 注册错误处理器
    @app.errorhandler(404)
    def not_found_error(error):
        return render_template('errors/404.html'), 404

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

    # 创建数据库表(如果不存在)
    with app.app_context():
        db.create_all()

    return app

6. 运行应用

# run.py
from app import create_app
import os

app = create_app()

if __name__ == '__main__':
    # 根据环境变量选择配置
    env = os.environ.get('FLASK_ENV', 'development')

    if env == 'production':
        app.run(host='0.0.0.0', port=5000, debug=False)
    else:
        app.run(host='127.0.0.1', port=5000, debug=True)
# config.py
import os
from datetime import timedelta

class Config:
    """基础配置"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///app.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    # 站点配置
    SITE_NAME = '我的Flask博客'
    POSTS_PER_PAGE = 10
    COMMENTS_PER_PAGE = 20

    # 安全配置
    SESSION_COOKIE_SECURE = False
    REMEMBER_COOKIE_DURATION = timedelta(days=7)

    # 文件上传配置
    UPLOAD_FOLDER = 'uploads'
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB

    # 邮件配置
    MAIL_SERVER = os.environ.get('MAIL_SERVER')
    MAIL_PORT = int(os.environ.get('MAIL_PORT') or 25)
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS') is not None
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    ADMINS = ['admin@example.com']

class DevelopmentConfig(Config):
    """开发环境配置"""
    DEBUG = True
    SQLALCHEMY_ECHO = True
    EXPLAIN_TEMPLATE_LOADING = False

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

class ProductionConfig(Config):
    """生产环境配置"""
    DEBUG = False
    SQLALCHEMY_ECHO = False
    SESSION_COOKIE_SECURE = True

    # 生产环境数据库配置
    SQLALCHEMY_POOL_SIZE = 20
    SQLALCHEMY_POOL_RECYCLE = 3600
    SQLALCHEMY_MAX_OVERFLOW = 30

7. 蓝图高级特性

7.1 嵌套蓝图

# api/__init__.py
from flask import Blueprint

api_bp = Blueprint('api', __name__, url_prefix='/api')

# 创建嵌套蓝图
from .v1 import v1_bp
from .v2 import v2_bp

# 注册嵌套蓝图
api_bp.register_blueprint(v1_bp, url_prefix='/v1')
api_bp.register_blueprint(v2_bp, url_prefix='/v2')

# api/v1/__init__.py
from flask import Blueprint

v1_bp = Blueprint('v1', __name__)

@v1_bp.route('/users')
def get_users():
    """API v1 获取用户列表"""
    return {'version': 'v1', 'endpoint': '/users'}

@v1_bp.route('/posts')
def get_posts():
    """API v1 获取文章列表"""
    return {'version': 'v1', 'endpoint': '/posts'}

# api/v2/__init__.py
from flask import Blueprint, jsonify

v2_bp = Blueprint('v2', __name__)

@v2_bp.route('/users')
def get_users_v2():
    """API v2 获取用户列表(增强版)"""
    return jsonify({
        'version': 'v2',
        'endpoint': '/users',
        'features': ['pagination', 'filtering', 'sorting']
    })

7.2 蓝图模板覆盖

# 蓝图模板优先级:蓝图模板 > 应用模板
# 如果蓝图有自己的模板,会优先使用蓝图的模板

# 示例:blog/templates/blog/base.html
# 会覆盖 app/templates/blog/base.html

# 在蓝图中使用应用级别的模板
@blog_bp.route('/about')
def about():
    """关于页面 - 使用应用级别的模板"""
    return render_template('about.html')  # 从应用模板目录查找

# 在蓝图中使用蓝图级别的模板
@blog_bp.route('/post/<slug>')
def post_detail(slug):
    """文章详情 - 使用蓝图模板"""
    return render_template('blog/post.html')  # 从蓝图模板目录查找

7.3 蓝图静态文件

<!-- 在模板中引用蓝图的静态文件 -->

<!-- 方法1:使用蓝图名称 -->
<link href="{{ url_for('blog.static', filename='css/blog.css') }}" rel="stylesheet">

<!-- 方法2:使用static端点(自动选择) -->
<link href="{{ url_for('static', filename='css/blog.css') }}" rel="stylesheet">

<!-- 蓝图静态文件目录结构 -->
blog/
    static/
        css/
            blog.css
        js/
            blog.js
        images/
            logo.png

7.4 蓝图请求钩子

@auth_bp.before_request
def before_auth_request():
    """认证蓝图请求前的处理"""
    # 检查用户是否已登录
    if request.endpoint not in ['auth.login', 'auth.register', 'auth.forgot_password']:
        if not current_user.is_authenticated:
            return redirect(url_for('auth.login'))

@auth_bp.after_request
def after_auth_request(response):
    """认证蓝图请求后的处理"""
    # 添加安全头
    response.headers['X-Content-Type-Options'] = 'nosniff'
    return response

@auth_bp.teardown_request
def teardown_auth_request(exception=None):
    """认证蓝图请求结束时的清理"""
    # 清理资源
    pass

# 蓝图级别的错误处理器
@auth_bp.errorhandler(404)
def auth_404_error(error):
    """认证模块的404错误处理"""
    return render_template('auth/404.html'), 404

8. 大型项目组织架构

# 大型Flask应用目录结构示例
myapp/
├── app/                          # 应用包
│   ├── __init__.py              # 应用工厂
│   ├── models.py                # 数据模型
│   ├── extensions.py            # 扩展实例
│   ├── commands.py              # CLI命令
│   ├── utils.py                 # 工具函数
│   ├── errors.py                # 错误处理器
│   ├── decorators.py            # 装饰器
│   ├── middleware.py            # 中间件
│   ├── auth/                    # 认证模块
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   ├── forms.py
│   │   ├── models.py
│   │   ├── utils.py
│   │   ├── templates/
│   │   │   └── auth/
│   │   │       ├── login.html
│   │   │       └── register.html
│   │   └── static/
│   │       ├── css/
│   │       └── js/
│   ├── blog/                    # 博客模块
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   ├── forms.py
│   │   ├── models.py
│   │   ├── templates/
│   │   │   └── blog/
│   │   │       ├── index.html
│   │   │       └── post.html
│   │   └── static/
│   ├── api/                     # API模块
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── posts.py
│   │   └── v2/
│   ├── admin/                   # 管理后台
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   └── templates/
│   │       └── admin/
│   ├── main/                    # 主模块(首页、关于等)
│   │   ├── __init__.py
│   │   └── routes.py
│   ├── static/                  # 全局静态文件
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   ├── templates/               # 全局模板
│   │   ├── base.html
│   │   ├── layout.html
│   │   └── errors/
│   └── tests/                   # 测试文件
│       ├── __init__.py
│       ├── test_auth.py
│       └── test_blog.py
├── migrations/                  # 数据库迁移文件
├── config.py                   # 配置文件
├── requirements.txt            # 依赖文件
├── .env                       # 环境变量
├── .gitignore                 # Git忽略文件
├── run.py                     # 启动文件
├── wsgi.py                    # WSGI入口
├── Dockerfile                 # Docker配置
└── docker-compose.yml         # Docker Compose配置

9. 最佳实践

  • 按功能模块划分蓝图:每个蓝图负责一个独立的功能模块
  • 使用应用工厂模式:方便配置、测试和扩展
  • 统一命名规范:保持蓝图名称、URL前缀的一致
  • 避免循环导入:在__init__.py中创建蓝图,在routes.py中导入
  • 使用上下文处理器:注入全局变量到所有模板
  • 配置管理:使用类管理不同环境的配置
  • 错误处理:在蓝图和应用级别都定义错误处理器
  • 测试友好:每个蓝图应有对应的测试文件

10. 常见问题解答

  • 小型项目:单一文件即可,不需要蓝图
  • 中型项目:路由超过20个,功能模块清晰时
  • 大型项目:必须使用蓝图进行模块化管理
  • 团队开发:多个开发者协作时
  • 需要复用:某些功能可能在其他项目中复用

# 创建支持子域名的蓝图
admin_bp = Blueprint(
    'admin',
    __name__,
    subdomain='admin',  # 指定子域名
    template_folder='templates'
)

# 注册蓝图时配置服务器名称
app = Flask(__name__)
app.config['SERVER_NAME'] = 'example.com'  # 设置主域名
app.register_blueprint(admin_bp)

# 访问地址:http://admin.example.com/

  1. 应用上下文:使用current_app.config
  2. 数据库:共享数据模型
  3. 会话:使用Flask会话
  4. 全局变量:在app/__init__.py中定义
  5. 配置文件:统一配置管理
  6. 消息队列:对于异步任务