当Flask应用变得复杂时,将所有代码放在一个文件中会导致:
蓝图解决了这些问题,它允许将应用分解为多个模块,每个模块可以独立开发。
# ❌ 不使用蓝图的应用结构(单个文件)
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
# 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
# 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
# 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)
# 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')
# 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
# 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
# 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']
})
# 蓝图模板优先级:蓝图模板 > 应用模板
# 如果蓝图有自己的模板,会优先使用蓝图的模板
# 示例: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') # 从蓝图模板目录查找
<!-- 在模板中引用蓝图的静态文件 -->
<!-- 方法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
@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
# 大型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配置
__init__.py中创建蓝图,在routes.py中导入
# 创建支持子域名的蓝图
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/
current_app.configapp/__init__.py中定义