第五十二篇:Web 框架与模板
前言
上一篇文章我们从零手写了一个 WSGI 应用和迷你框架。虽然体验了底层原理,但实际开发中我们不会每次都从 WSGI 开始。Python 生态中有许多成熟的 Web 框架,它们封装了 WSGI、路由、模板等常用功能。本文将介绍 Flask 框架的快速入门、Jinja2 模板引擎语法,以及 MVC 设计思想。
Flask 快速入门
为什么选 Flask
┌──────────────┬───────────────┬──────────────────┐
│ Django │ Flask │ FastAPI │
├──────────────┼───────────────┼──────────────────┤
│ 全栈框架 │ 微框架 │ ASGI 异步框架 │
│ 内置 ORM │ 按需扩展 │ 自动 API 文档 │
│ 内置 Admin │ 灵活轻量 │ 类型提示优先 │
│ 适合大型项目 │ 适合小型/API │ 适合高性能 API │
│ 学习曲线陡 │ 学习曲线平缓 │ 中等 │
└──────────────┴───────────────┴──────────────────┘
Flask 的核心优势:
- 微型框架:核心简单,按需选择扩展库
- 灵活自由:不强制目录结构和数据库选择
- 强大生态:flask-login、flask-sqlalchemy 等丰富扩展
- 易于理解:源码仅数千行,适合学习框架原理
- 社区活跃:大量教程和第三方扩展
安装与第一个 Flask 应用
pip install flask
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return '<h1>Hello, Flask!</h1>'
@app.route('/about')
def about():
return '<h2>About</h2><p>This is the about page.</p>'
if __name__ == '__main__':
app.run(debug=True)
路由系统详解
from flask import Flask
app = Flask(__name__)
# ========== 基本路由 ==========
@app.route('/')
def index():
return '首页'
# ========== 动态参数路由 ==========
@app.route('/users/<int:user_id>')
def get_user(user_id):
return f'<h3>用户 {user_id}</h3>'
@app.route('/posts/<path:post_slug>')
def get_post(post_slug):
# path: 支持多级路径,如 posts/2024/01/my-post
return f'<h3>文章: {post_slug}</h3>'
# ========== URL 构建 ==========
@app.route('/profile/<username>')
def profile(username):
from flask import url_for
# url_for 生成安全 URL(防硬编码)
logout_url = url_for('logout', _external=True)
return f'<a href="{logout_url}">退出登录</a>'
@app.route('/logout')
def logout():
return '已退出'
# ========== HTTP 方法限制 ==========
@app.route('/api/data', methods=['GET'])
def get_data():
return '{"message": "GET request"}'
@app.route('/api/data', methods=['POST'])
def create_data():
return '{"status": "created"}', 201
# 简写形式
@app.get('/api/items')
def list_items():
return '{"items": []}'
@app.post('/api/items')
def add_item():
return '{"id": 1}', 201
@app.put('/api/items/<int:item_id>')
def update_item(item_id):
return f'{{"updated": {item_id}}}'
@app.delete('/api/items/<int:item_id>')
def delete_item(item_id):
return f'{{"deleted": {item_id}}}'
# ========== 自定义转换器 ==========
class ListConverter(Flask.converters.BaseConverter):
"""自定义 URL 转换器:支持列表参数 /items/1,2,3"""
def to_python(self, value):
return [int(x) for x in value.split(',')]
def to_url(self, value):
return ','.join(str(v) for v in value)
app.url_map.converters['list'] = ListConverter
@app.route('/items/<list:item_ids>')
def get_items(item_ids):
return f'<p>商品 ID 列表: {item_ids}</p>'
# 使用示例:访问 /items/1,2,3 → item_ids = [1, 2, 3]
请求与响应处理
from flask import Flask, request, jsonify, make_response
app = Flask(__name__)
# ========================================
# 请求处理
# ========================================
@app.get('/query-params')
def handle_query():
# 查询参数 ?name=John&age=25
name = request.args.get('name', 'World')
age = request.args.get('age', 0, type=int)
return f'你好, {name}! 年龄: {age}'
@app.get('/headers-demo')
def show_headers():
headers = dict(request.headers)
useful = {k: v for k, v in headers.items()
if not k.startswith('_')}
return jsonify(useful)
@app.post('/form-data')
def handle_form():
username = request.form.get('username')
email = request.form.get('email')
if not username or not email:
return jsonify({'error': '缺少必填字段'}), 400
return jsonify({
'success': True,
'username': username,
'email': email
})
@app.post('/json-data')
def handle_json():
data = request.get_json()
if not data:
return jsonify({'error': '无效的 JSON'}), 400
return jsonify({'received': data})
@app.post('/upload-file')
def upload_file():
file = request.files.get('file')
if not file:
return jsonify({'error': '未找到文件'}), 400
filename = file.filename
file.save(f'/tmp/uploads/{filename}')
return jsonify({
'success': True,
'filename': filename,
'size': file.content_length
})
@app.get('/path-params/<int:user_id>/<string:action>')
def user_action(user_id, action):
return f'用户 {user_id} 执行操作: {action}'
# ========================================
# 响应处理
# ========================================
@app.get('/response-types')
def response_types():
return {'message': 'Custom headers'}, 201, {
'X-Custom-Header': 'Flask',
'X-Powered-By': 'Python'
}
@app.get('/json-response')
def json_response():
return jsonify({
'code': 0,
'data': {'name': 'Python', 'version': '3.12'},
'msg': '成功'
})
@app.get('/html-response')
def html_response():
html = '''
<h1 style="color: #654ACB;">Flask HTML 响应</h1>
<p>这是由 Flask 直接返回的 HTML 内容。</p>
'''
return html
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Not Found', 'message': str(error)}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({'error': 'Server Error'}), 500
视图函数进阶
基于类的视图(Class-Based Views)
from flask import Flask, views
class ResourceView(views.MethodView):
def get(self, **kwargs):
return {'method': 'GET', 'params': kwargs}
def post(self, **kwargs):
return {'method': 'POST', 'params': kwargs}
app.add_url_rule('/api/users', view_func=ResourceView.as_view('users'))
Before/After Request 钩子
from flask import Flask, g, request
import time
@app.before_request
def auth_check():
token = request.headers.get('Authorization', '')
if token and len(token) > 10:
g.user_token = token
else:
return jsonify({'error': '未授权'}), 401
@app.after_request
def add_security_headers(response):
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-Content-Type-Options'] = 'nosniff'
return response
Flask 配置管理
from flask import Flask
app = Flask(__name__)
# 直接设置
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'super-secret-key'
# 环境变量
app.config.from_prefixed_env() # FLASK_SECRET_KEY=mykey
# 配置文件
app.config.from_object('config.DevelopmentConfig')
# 工厂函数
def create_app(config_name=None):
app = Flask(__name__)
app.config.from_object(config_name or 'config.DefaultConfig')
return app
Jinja2 模板引擎
为什么要用模板
❌ 不用模板:所有逻辑混在视图代码中
return f"""
<html><body><h1>{title}</h1>
{% for item in items %}<li>{item.name}</li>{% endfor %}
</body></html>
"""
✅ 使用模板:HTML 文件中直接写模板语法
<h1>{{ title }}</h1>
{% for item in items %}
<li>{{ item.name }}</li>
{% endfor %}
基础语法
from flask import Flask, render_template
@app.route('/')
def home():
context = {
'title': '我的博客首页',
'user': {'name': '张三', 'age': 28, 'is_admin': True},
'posts': [
{'id': 1, 'title': 'Python 入门', 'views': 100},
{'id': 2, 'title': 'Web 开发指南', 'views': 200},
],
}
return render_template('home.html', **context)
模板文件 templates/home.html
<h1>{{ title }}</h1>
<p>欢迎, {{ user.name }}!</p>
{% if user.is_admin %}
<span style="color: red;">管理员</span>
{% endif %}
<p>简介: {{ user.bio | default('暂无简介') }}</p>
<ul>
{% for post in posts %}
<li><strong>{{ post.title }}</strong> - {{ post.views }} 浏览</li>
{% endfor %}
</ul>
<p>用户名: {{ user.name | upper }}</p>
<p>年份: {{ current_year }}</p>
常用过滤器速查表
| 过滤器 | 用法 | 说明 |
|---|---|---|
default |
{{ value | default('N/A') }} |
空值时显示默认值 |
upper/lower |
{{ name | upper }} |
大小写转换 |
truncate |
{{ text | truncate(30) }} |
截断到指定长度 |
join |
{{ tags | join(', ') }} |
连接为字符串 |
safe |
{{ html | safe }} |
标记为安全的 HTML |
round |
{{ num | round(1) }} |
四舍五入 |
format |
{{ '%s: %d' | format(name, count) }} |
字符串格式化 |
模板继承机制
base.html ──→ child.html
{% block content %} {% extends 'base.html' %}
... {% block content %}
{% endblock %} ...
{% endblock %}
父模板 templates/base.html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{% block title %}我的网站{% endblock %}</title>
<style>
body { font-family: sans-serif; margin: 0; }
nav { background: #654ACB; padding: 15px; color: white; }
.container { max-width: 900px; margin: 20px auto; }
</style>
</head>
<body>
<nav>
<a href="/">首页</a>
<a href="/posts">文章</a>
</nav>
<div class="container">
{% block content %}{% endblock %}
</div>
</body>
</html>
子模板 templates/post.html
{% extends 'base.html' %}
{% block title %}{{ post.title }} - 我的网站{% endblock %}
{% block content %}
<article>
<h1>{{ post.title }}</h1>
<div>{{ post.content | nl2br | safe }}</div>
<h3>评论</h3>
<ul>
{% for comment in comments %}
<li><strong>{{ comment.author }}</strong>: {{ comment.text }}</li>
{% else %}
<li>暂无评论</li>
{% endfor %}
</ul>
</article>
{% endblock %}
宏(Macro)—— 模板中的函数
{% macro input_field(name, value='', placeholder='') %}
<input type="text" name="{{ name }}"
value="{{ value }}"
placeholder="{{ placeholder }}"/>
{% endmacro %}
{% macro card(title, content) %}
<div class="card">
<h3>{{ title }}</h3>
<p>{{ content }}</p>
</div>
{% endmacro %}
模板控制流
{% for item in items %}
<tr>
<td>{{ loop.index }}</td>
<td>{{ item.name }}</td>
<td>
{% if item.active %}
<span>✓ 在线</span>
{% else %}
<span>✗ 离线</span>
{% endif %}
</td>
</tr>
{% else %}
<tr><td colspan="3">没有数据</td></tr>
{% endfor %}
<!-- 特殊循环变量 -->
{% for item in items %}
序号: {{ loop.index }} | 总数: {{ loop.length }} |
索引: {{ loop.index0 }} | 奇偶: {{ '偶数' if loop.even else '奇数' }}
{% endfor %}
MVC 设计思想
MVC 架构模式
View (视图)
Jinja2 模板 / HTML / JSON
↓
Model (模型)
SQLAlchemy ORM / 业务逻辑
↓
Controller (控制器)
Flask 路由 / 视图函数
← 用户 → 浏览器 → Controller → Model → View → 响应
Flask 中实现 MVC 风格
from flask import Flask, Blueprint
app = Flask(__name__)
# M - Model 层
class PostModel:
def __init__(self):
self.posts = [{'id': 1, 'title': 'Hello', 'views': 10}]
def get_all(self):
return sorted(self.posts, key=lambda p: p['id'], reverse=True)
def get_by_id(self, pid):
return next((p for p in self.posts if p['id'] == pid), None)
def create(self, data):
new_id = max(p['id'] for p in self.posts) + 1
post = {'id': new_id, **data}
self.posts.append(post)
return post
model = PostModel()
# C - Controller 层(蓝图)
posts_bp = Blueprint('posts', __name__)
@posts_bp.get('/')
def list_posts():
return render_template('posts/list.html', posts=model.get_all())
@posts_bp.get('/<int:post_id>')
def view_post(post_id):
post = model.get_by_id(post_id)
if not post:
return jsonify({'error': 'Not Found'}), 404
return render_template('posts/detail.html', post=post)
@posts_bp.post('/')
def create_post():
data = request.get_json()
new_post = model.create(data)
return jsonify(new_post), 201
app.register_blueprint(posts_bp, url_prefix='/posts')
蓝图的模块化组织
project/
├── app.py # 应用入口
├── routes/
│ ├── api.py # API 路由蓝图
│ └── web.py # 网页路由蓝图
├── models/ # 数据模型
├── templates/ # 模板
└── static/ # 静态文件
from .api import api_bp
from .web import web_bp
def register_blueprints(app):
app.register_blueprint(api_bp, url_prefix='/api')
app.register_blueprint(web_bp)
实战项目:完整博客系统
from flask import Flask, render_template, request, jsonify, flash, redirect, url_for, abort
from datetime import datetime
app = Flask(__name__)
app.secret_key = 'dev-secret-key-change-in-production'
class BlogData:
"""模拟数据库"""
def __init__(self):
self.categories = [
{'id': 1, 'name': '技术'},
{'id': 2, 'name': '生活'},
]
self.posts = [
{
'id': 1,
'title': 'Python Web 开发完全指南',
'content': '本文介绍如何使用 Python 进行 Web 开发...',
'category_id': 1,
'views': 1000,
'created_at': datetime.now().strftime('%Y-%m-%d'),
'status': 'published',
},
{
'id': 2,
'title': 'Docker 容器化部署实践',
'content': 'Docker 是现代 DevOps 的核心工具...',
'category_id': 1,
'views': 800,
'created_at': datetime.now().strftime('%Y-%m-%d'),
'status': 'published',
},
]
self.comments = [
{'id': 1, 'post_id': 1, 'author': 'reader1',
'content': '好文章!', 'created_at': datetime.now().strftime('%Y-%m-%d')},
]
def get_published_posts(self, page=1, per_page=10):
posts = [p for p in self.posts if p['status'] == 'published']
posts.sort(key=lambda x: x['id'], reverse=True)
total = len(posts)
start = (page - 1) * per_page
return posts[start:start+per_page], total
def get_post(self, post_id):
post = next((p for p in self.posts if p['id'] == post_id), None)
if post:
post['views'] += 1
return post
def get_comments(self, post_id):
return [c for c in self.comments if c['post_id'] == post_id]
def add_comment(self, post_id, author, content):
comment = {
'id': max(c['id'] for c in self.comments) + 1,
'post_id': post_id, 'author': author,
'content': content,
'created_at': datetime.now().strftime('%Y-%m-%d'),
}
self.comments.append(comment)
return comment
data = BlogData()
@app.get('/')
def index():
page = request.args.get('page', 1, type=int)
posts, total = data.get_published_posts(page=page)
total_pages = (total + 9) // 10
return render_template('index.html',
posts=posts, total_pages=total_pages)
@app.get('/post/<int:post_id>')
def post_detail(post_id):
post = data.get_post(post_id)
if not post:
abort(404)
comments = data.get_comments(post_id)
related = [p for p in data.posts
if p['id'] != post_id
and p['category_id'] == post['category_id']][:3]
return render_template('post.html', post=post, comments=comments, related=related)
@app.post('/post/<int:post_id>/comment')
def add_comment(post_id):
content = request.form.get('content', '').strip()
if not content:
flash('评论内容不能为空', 'warning')
return redirect(url_for('post_detail', post_id=post_id))
author = request.form.get('author', '匿名').strip() or '匿名'
data.add_comment(post_id, author, content)
flash('评论发表成功!', 'success')
return redirect(url_for('post_detail', post_id=post_id))
@app.get('/api/posts')
def api_posts():
page = request.args.get('page', 1, type=int)
posts, total = data.get_published_posts(page=page)
return jsonify({'code': 0, 'data': posts, 'total': total})
@app.errorhandler(404)
def handle_404(e):
return jsonify({'error': 'Not Found', 'message': str(e)}), 404
if __name__ == '__main__':
print("博客系统运行中...")
app.run(debug=True)
小结
| 知识点 | 说明 |
|---|---|
| Flask | 微型 Web 框架,核心简单灵活 |
| 路由系统 | @app.route() 装饰器、URL 参数 <id>、HTTP 方法限制 |
| URL 构建 | url_for() 替代硬编码 URL |
| 请求处理 | request.args/form/json/files 四种数据来源 |
| 响应处理 | 字符串 / JSON / HTML / 重定向 / 元组 (body, code, headers) |
| 视图模式 | 函数视图 vs 类视图 (MethodView) |
| 请求钩子 | before_request / after_request / teardown_appcontext |
| 模板继承 | {% extends %} + {% block %} 实现布局复用 |
| Jinja2 过滤器 | 丰富的内置过滤器,链式使用 |
| 宏(Macro) | 模板中的函数,减少重复代码 |
| MVC 分层 | Model(数据) → Controller(逻辑) → View(展示) |
| 蓝图(Blueprint) | 模块化组织大型项目的路由和模板 |
核心原则:Flask 遵循"约定优于配置"的最小哲学,它不强制你的项目结构,而是提供足够的灵活性让你自由组合。学习 Flask 后,理解 Django、FastAPI 等其他框架会非常容易——它们的本质都是 WSGI/ASGI 之上的不同抽象层。
练习题
为博客系统添加分类筛选功能:在首页 URL 中添加
/category/<category_id>路由,只显示该分类下的文章,并在侧边栏列出所有分类及其文章数量。实现一个简单的缓存中间件:对
/api/posts接口的结果缓存 60 秒,期间同一接口的请求直接返回缓存数据。提示:使用functools.lru_cache或 Redis。将博客系统的模板从内联字符串改为外部模板文件(放在
templates/目录下),并使用{% extends %}和{% block %}实现布局继承。尝试为首页、详情页、错误页面分别创建模板。
