章节 1：Flask 是个啥？为什么它这么香？
Flask 是一个 Python 的微框架，诞生于 2010 年，由 Armin Ronacher 开发。所谓“微”，并不是说它功能少，而是指它的核心代码极简，只提供最基本的功能，比如路由、请求处理和响应，让你有无限的自由度去扩展。相比 Django 那样的“全家桶”框架，Flask 更像一个轻巧的工具箱，你可以根据项目需求随意挑选工具。
Flask 的核心优势
轻量灵活：核心代码不到 1 万行，安装包只有几十 KB，启动一个 Web 应用只需要几行代码。
扩展生态丰富：想加数据库？用 Flask-SQLAlchemy。需要用户认证？试试 Flask-Login。几乎每个常见功能都有成熟的扩展库。
学习曲线平滑：如果你会 Python，学 Flask 几乎是零成本，写起来像写普通的 Python 函数。
完全掌控：不像某些框架强制你用特定的目录结构，Flask 让你自由组织代码，适合小项目，也能扩展到大项目。
适合 Flask 的场景
Flask 特别适合以下项目：
快速原型开发：想验证一个想法？几小时就能搭出 MVP。
RESTful API：用 Flask 写接口简直如丝般顺滑。
小型到中型 Web 应用：博客、论坛、任务管理工具等。
定制化需求强的项目：你需要完全控制应用的每个细节。
但如果你需要一个开箱即用的后台管理系统，或者项目规模特别大（比如上百个模型），可能 Django 会更省心。不过，Flask 的灵活性让它在许多场景下依然是王者！
小试牛刀：感受 Flask 的简洁
来，写几行代码，感受一下 Flask 的魅力：
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello, Flask!'
if __name__ == '__main__':
app.run(debug=True)保存为 app.py，然后在终端运行 python app.py，打开浏览器访问 http://127.0.0.1:5000，你会看到页面显示 Hello, Flask!。就这么简单，一个 Web 应用已经跑起来了！这几行代码包含了 Flask 的核心：创建应用、定义路由、返回响应。
章节 2：环境搭建，迈出第一步
在动手写代码之前，咱们得先把开发环境准备好。
安装 Python
Flask 需要 Python 3.6 或以上版本。建议用最新稳定版（比如 3.10 或 3.11）。如果你还没装 Python，去 python.org 下载安装包，按提示安装。安装后，在终端运行：
python --version如果显示类似 Python 3.10.4，说明 Python 已经就位。如果命令不行，试试 python3 --version（Linux/macOS 可能需要用 python3）。
创建虚拟环境
虚拟环境是必须的！ 它能隔离项目依赖，避免不同项目之间包版本冲突。Python 自带 venv 模块，超级好用。假设你想在 my_flask_project 文件夹下开发，先创建项目目录：
mkdir my_flask_project
cd my_flask_project然后创建虚拟环境：
python -m venv venv这会在项目目录下生成一个 venv 文件夹。激活虚拟环境：
Windows：venv\Scripts\activate
macOS/Linux：source venv/bin/activate
激活后，终端提示符会变（比如前面多了 (venv)），说明你进入了虚拟环境。现在，安装的包只会影响这个项目，不会污染全局环境。
安装 Flask
在虚拟环境中，运行：
pip install flask这会安装最新版的 Flask 和它的依赖（比如 Werkzeug、Jinja2）。检查安装是否成功：
pip show flask会显示 Flask 的版本信息，比如 Version: 2.3.2。如果没问题，环境就搞定了！
开发工具推荐
代码编辑器：VS Code 是不二之选，支持 Python 插件、调试、代码补全。PyCharm 也很强，尤其是专业版，但免费版够用了。
终端：Windows 用 PowerShell 或 Git Bash，macOS/Linux 自带终端就很好。
浏览器开发者工具：Chrome 或 Firefox 的 F12 开发者工具，调试前端和网络请求必备。
章节 3：打造你的第一个 Flask 应用
现在，我们要用 Flask 建一个简单的 Web 应用，带你理解 Flask 的核心组件。这个应用会有几个页面，比如首页、关于页面，还能处理用户输入。
项目结构
先规划一下项目目录，保持代码整洁：
my_flask_project/
├── app.py
├── templates/
│ └── index.html
├── static/
│ └── style.css
└── venv/app.py：主应用文件，包含 Flask 代码。
templates/：存放 HTML 模板。
static/：放 CSS、JS 等静态文件。
venv/：虚拟环境文件夹（已创建）。
编写核心代码
在 app.py 中写以下代码：
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/')
def home():
return render_template('index.html', title='欢迎体验 Flask')
@app.route('/about')
def about():
return render_template('about.html', title='关于我们')
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)这段代码做了啥？
导入了 Flask 和 render_template（用于渲染 HTML 模板）。
创建了 Flask 应用实例 app。
定义了两个路由：/（首页）和 /about（关于页面）。
使用 render_template 渲染对应的 HTML 文件，并传递一个 title 参数。
运行应用，debug=True 开启调试模式，代码改动会自动重载；host='0.0.0.0' 允许局域网访问。
创建 HTML 模板
Flask 默认使用 Jinja2 模板引擎，模板文件放进 templates 文件夹。创建 templates/index.html：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ title }}</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<h1>{{ title }}</h1>
<p>这是一个用 Flask 构建的简单 Web 应用！</p>
<a href="{{ url_for('about') }}">去看看关于页面</a>
</body>
</html>再创建 templates/about.html：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{{ title }}</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<h1>{{ title }}</h1>
<p>我们是一个热爱 Flask 的团队，致力于分享 Web 开发知识！</p>
<a href="{{ url_for('home') }}">返回首页</a>
</body>
</html>注意：
{{ title }} 是 Jinja2 语法，渲染 Python 传来的变量。
{{ url_for('static', filename='style.css') }} 生成静态文件路径，防止硬编码。
{{ url_for('about') }} 和 {{ url_for('home') }} 生成路由的 URL，动态适应路由变化。
添加 CSS 样式
在 static 文件夹下创建 style.css：
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
background-color: #f9f9f9;
}
h1 {
color: #2c3e50;
}
a {
color: #3498db;
text-decoration: none;
}
a:hover {
text-decoration: underline;
}运行应用
确保在虚拟环境中，运行：
python app.py打开浏览器，访问 http://127.0.0.1:5000，你会看到一个简单的首页，点击链接可以跳转到关于页面。是不是超有成就感？
调试技巧
调试模式：debug=True 会显示详细错误信息，代码改动自动重载。但上线时必须关闭，否则有安全风险。
日志查看：Flask 默认把日志输出到终端，错误信息会直接显示。
浏览器检查：用 F12 查看网络请求和 HTML 结构，确保静态文件加载正常。
这个小应用已经展示了 Flask 的核心功能：路由、模板、静态文件。
章节 4：路由与模板的进阶技巧
Flask 的路由和模板是构建动态 Web 应用的基础。
路由的魔法
Flask 的路由通过 @app.route() 装饰器定义，映射 URL 到 Python 函数。咱们先从基础用法开始，逐步解锁高级技巧。
动态路由
假设你想根据用户名显示用户主页，比如 /user/张三 显示张三的信息。Flask 支持动态路由，用 <变量名> 占位：
@app.route('/user/<username>')
def user_profile(username):
return render_template('user.html', username=username)创建 templates/user.html：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>用户主页</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<h1>欢迎，{{ username }}！</h1>
<p>这是你的个人主页。</p>
<a href="{{ url_for('home') }}">返回首页</a>
</body>
</html>访问 http://127.0.0.1:5000/user/张三，页面会显示“欢迎，张三！”。<username> 会捕获 URL 中的值，传给函数参数。
指定参数类型
默认情况下，<username> 捕获字符串。如果你想要数字 ID，比如 /user/123，可以用类型转换器 <int:id>：
@app.route('/user/<int:id>')
def user_by_id(id):
return f'用户 ID: {id}'访问 /user/123 显示“用户 ID: 123”，但 /user/abc 会返回 404 错误。支持的类型包括：
string：默认，任意字符串（不含斜杠）。
int：整数。
float：浮点数。
path：类似字符串，但允许包含斜杠。
uuid：UUID 字符串。
多参数路由
可以同时捕获多个参数，比如 /post/2023/10：
@app.route('/post/<int:year>/<int:month>')
def post_by_date(year, month):
return f'文章归档：{year}年{month}月'访问 /post/2023/10 显示“文章归档：2023年10月”。
查询参数
URL 里的 ?key=value 叫查询参数，比如 http://127.0.0.1:5000/search?q=flask。用 request.args 获取：
from flask import request
@app.route('/search')
def search():
query = request.args.get('q', '') # 如果 q 不存在，返回空字符串
return f'搜索关键词：{query}'访问 /search?q=flask 显示“搜索关键词：flask”。request.args.get() 是安全的方式，避免键不存在时抛异常。
Jinja2 模板的超能力
Jinja2 是 Flask 默认的模板引擎，功能强大又易用。咱们来探索几个高级用法，让页面更动态！
模板继承
多个页面通常有相同的头部、导航栏、底部，重复写这些代码太烦了。Jinja2 支持模板继承，让你定义一个基础模板，其他页面继承它。
创建 templates/base.html：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{% block title %}{% endblock %}</title>
<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
<nav>
<a href="{{ url_for('home') }}">首页</a> |
<a href="{{ url_for('about') }}">关于</a>
</nav>
<div class="content">
{% block content %}
{% endblock %}
</div>
<footer>
<p>© 2025 Flask 教程</p>
</footer>
</body>
</html>然后修改 templates/index.html：
{% extends 'base.html' %}
{% block title %}{{ title }}{% endblock %}
{% block content %}
<h1>{{ title }}</h1>
<p>这是一个用 Flask 构建的简单 Web 应用！</p>
<a href="{{ url_for('about') }}">去看看关于页面</a>
{% endblock %}{% extends 'base.html' %} 表示继承 base.html，{% block %} 填充父模板的占位块。这样，index.html 只需写独特的内容，公共部分由 base.html 提供。about.html 和 user.html 也类似改一下。
循环和条件
Jinja2 支持 Python 风格的循环和条件判断。比如，显示一个用户列表：
@app.route('/users')
def user_list():
users = ['张三', '李四', '王五']
return render_template('users.html', users=users)创建 templates/users.html：
{% extends 'base.html' %}
{% block title %}用户列表{% endblock %}
{% block content %}
<h1>用户列表</h1>
{% if users %}
<ul>
{% for user in users %}
<li>{{ user }}</li>
{% endfor %}
</ul>
{% else %}
<p>暂无用户！</p>
{% endif %}
{% endblock %}访问 /users，会显示一个用户列表。如果 users 为空，显示“暂无用户！”。
过滤器
Jinja2 提供过滤器，转换变量输出。比如，把用户名大写：
<p>用户名：{{ username | upper }}</p>常见过滤器：
upper：大写。
lower：小写。
truncate(length)：截断字符串。
safe：渲染 HTML 内容（小心 XSS 攻击）。
default('默认值')：变量不存在时用默认值。
表单处理
用户输入是 Web 应用的核心。来建一个简单的搜索表单：
@app.route('/search', methods=['GET', 'POST'])
def search_form():
if request.method == 'POST':
query = request.form.get('query')
return f'你搜索了：{query}'
return render_template('search.html')创建 templates/search.html：
{% extends 'base.html' %}
{% block title %}搜索{% endblock %}
{% block content %}
<h1>搜索</h1>
<form method="POST">
<input type="text" name="query" placeholder="输入关键词">
<button type="submit">搜索</button>
</form>
{% endblock %}methods=['GET', 'POST'] 允许路由处理 GET 和 POST 请求。GET 显示表单，POST 处理提交数据。request.form.get('query') 获取表单字段。
章节 5：数据库集成，让数据活起来
一个真正的 Web 应用离不开数据存储，比如保存用户信息、文章内容或任务记录。Flask 本身不提供数据库功能，但通过扩展（比如 Flask-SQLAlchemy），可以轻松集成各种数据库。我们以 SQLite 作为示例（轻量、无需额外安装），但也会提到 MySQL 的配置方法。
为什么选择 Flask-SQLAlchemy？
Flask-SQLAlchemy 是 SQLAlchemy 的 Flask 封装，SQLAlchemy 是 Python 里最强大的 ORM（对象关系映射）工具。它能让你用 Python 类操作数据库，写起来就像操作对象一样直观。好处多多：
简单易用：定义模型就像写 Python 类。
跨数据库支持：支持 SQLite、MySQL、PostgreSQL 等。
事务管理：自动处理数据库连接和事务，省心省力。
查询灵活：链式查询，写复杂查询也不头疼。
安装依赖
在虚拟环境中，运行：
pip install flask-sqlalchemy如果想用 MySQL，还需要安装驱动：
pip install pymysql配置数据库
在 app.py 中添加数据库配置。SQLite 简单，直接用文件存储；MySQL 需要数据库服务支持。修改 app.py：
from flask import Flask, render_template
from flask_sqlalchemy import SQLAlchemy
import os
app = Flask(__name__)
# 配置 SQLite 数据库（文件存储在项目目录下）
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
# 禁用跟踪对象修改（节省性能）
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 初始化数据库
db = SQLAlchemy(app)
# 路由和模板（沿用之前的代码）
@app.route('/')
def home():
return render_template('index.html', title='欢迎体验 Flask')
@app.route('/about')
def about():
return render_template('about.html', title='关于我们')
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)注意：
sqlite:///site.db 表示数据库文件 site.db 保存在项目根目录。
如果用 MySQL，URI 格式是 mysql+pymysql://用户名:密码@主机/数据库名，比如 mysql+pymysql://root:password@localhost/myapp。
db = SQLAlchemy(app) 初始化数据库，连接 Flask 应用。
定义模型
模型是数据库表的 Python 表示。假设我们要建一个博客应用，存储用户和文章。添加以下模型到 app.py（稍后会重构代码结构）：
# 定义用户模型
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
posts = db.relationship('Post', backref='author', lazy=True)
def __repr__(self):
return f"User('{self.username}', '{self.email}')"
# 定义文章模型
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
def __repr__(self):
return f"Post('{self.title}')"解释：
db.Model 是基类，定义表结构。
db.Column 定义表字段，primary_key=True 表示主键，unique=True 确保唯一，nullable=False 不允许为空。
db.relationship 建立用户和文章的一对多关系，backref 让文章反向访问作者。
__repr__ 方法方便调试，打印对象时显示清晰信息。
创建数据库
在终端进入 Python 交互模式：
python
>>> from app import db
>>> db.create_all()
>>> exit()这会在项目目录下生成 site.db，包含 user 和 post 表。每次模型变更后，需重新运行 db.create_all()，但已有数据不会自动迁移（后面会讲迁移工具）。
增删改查
让我们实现文章的 CRUD（创建、读取、更新、删除）功能。更新 app.py，添加新路由：
from flask import request, redirect, url_for
@app.route('/posts', methods=['GET', 'POST'])
def posts():
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
user_id = 1 # 暂时硬编码，后面会用登录用户
post = Post(title=title, content=content, user_id=user_id)
db.session.add(post)
db.session.commit()
return redirect(url_for('posts'))
posts = Post.query.all()
return render_template('posts.html', posts=posts)创建 templates/posts.html：
{% extends 'base.html' %}
{% block title %}文章列表{% endblock %}
{% block content %}
<h1>所有文章</h1>
<form method="POST">
<input type="text" name="title" placeholder="文章标题" required>
<textarea name="content" placeholder="文章内容" required></textarea>
<button type="submit">发布文章</button>
</form>
<hr>
{% if posts %}
<ul>
{% for post in posts %}
<li><strong>{{ post.title }}</strong>: {{ post.content }}</li>
{% endfor %}
</ul>
{% else %}
<p>暂无文章，快来发布吧！</p>
{% endif %}
{% endblock %}访问 /posts，你可以看到一个表单，提交后文章会保存到数据库并显示在列表中。是不是超酷？
查询示例
获取所有文章：Post.query.all()
根据 ID 查找：Post.query.get(1)
过滤：Post.query.filter_by(user_id=1).all()
分页：Post.query.paginate(page=1, per_page=10)
小技巧与注意事项
事务管理：用 db.session.commit() 提交变更，db.session.rollback() 回滚错误。
性能优化：查询大数据时用 lazy='dynamic' 或分页，避免加载过多数据。
迁移工具：db.create_all() 不支持表结构变更，推荐用 Flask-Migrate 管理数据库迁移（后文会讲）。
现在你的应用已经能存储和展示数据了！接下来，我们让它支持用户注册和登录。
章节 6：用户认证，保护你的应用
用户认证是 Web 应用的核心功能，涉及注册、登录、会话管理和密码安全。这一章，我们用 Flask-Login 和 Flask-Bcrypt 实现安全的用户认证系统。
安装依赖
运行：
pip install flask-login flask-bcryptFlask-Login：管理用户会话，支持登录、登出、记住我功能。
Flask-Bcrypt：安全地哈希密码，防止明文存储。
配置 Flask-Login 和 Bcrypt
更新 app.py：
from flask import Flask, render_template, request, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required
from flask_bcrypt import Bcrypt
import os
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'your-secret-key' # 用于会话和 CSRF 保护
db = SQLAlchemy(app)
bcrypt = Bcrypt(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login' # 未登录时重定向到登录页面
# 用户模型（更新）
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password = db.Column(db.String(60), nullable=False)
posts = db.relationship('Post', backref='author', lazy=True)
def __repr__(self):
return f"User('{self.username}', '{self.email}')"
# 加载用户
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))UserMixin 提供 Flask-Login 所需的方法（如 is_authenticated）。
SECRET_KEY 用于加密会话，生产环境中用随机字符串（可用 os.urandom(24) 生成）。
login_manager.login_view 指定未登录时的重定向路由。
注册功能
添加注册路由和模板：
@app.route('/register', methods=['GET', 'POST'])
def register():
if request.method == 'POST':
username = request.form.get('username')
email = request.form.get('email')
password = request.form.get('password')
hashed_password = bcrypt.generate_password_hash(password).decode('utf-8')
user = User(username=username, email=email, password=hashed_password)
try:
db.session.add(user)
db.session.commit()
flash('注册成功！请登录', 'success')
return redirect(url_for('login'))
except:
db.session.rollback()
flash('用户名或邮箱已存在！', 'danger')
return render_template('register.html')创建 templates/register.html：
{% extends 'base.html' %}
{% block title %}注册{% endblock %}
{% block content %}
<h1>注册新用户</h1>
{% if messages %}
{% for message in messages %}
<p class="{{ 'text-success' if message.category == 'success' else 'text-danger' }}">{{ message.message }}</p>
{% endfor %}
{% endif %}
<form method="POST">
<input type="text" name="username" placeholder="用户名" required>
<input type="email" name="email" placeholder="邮箱" required>
<input type="password" name="password" placeholder="密码" required>
<button type="submit">注册</button>
</form>
<p>已有账号？<a href="{{ url_for('login') }}">去登录</a></p>
{% endblock %}bcrypt.generate_password_hash 加密密码，存储哈希值。
flash 显示一次性消息（成功或失败），需在模板中用 get_flashed_messages() 获取。
登录功能
添加登录路由：
from flask_login import current_user
@app.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('home'))
if request.method == 'POST':
email = request.form.get('email')
password = request.form.get('password')
user = User.query.filter_by(email=email).first()
if user and bcrypt.check_password_hash(user.password, password):
login_user(user, remember=request.form.get('remember'))
flash('登录成功！', 'success')
return redirect(url_for('home'))
else:
flash('邮箱或密码错误！', 'danger')
return render_template('login.html')创建 templates/login.html：
{% extends 'base.html' %}
{% block title %}登录{% endblock %}
{% block content %}
<h1>用户登录</h1>
{% if messages %}
{% for message in messages %}
<p class="{{ 'text-success' if message.category == 'success' else 'text-danger' }}">{{ message.message }}</p>
{% endfor %}
{% endif %}
<form method="POST">
<input type="email" name="email" placeholder="邮箱" required>
<input type="password" name="password" placeholder="密码" required>
<label><input type="checkbox" name="remember"> 记住我</label>
<button type="submit">登录</button>
</form>
<p>没有账号？<a href="{{ url_for('register') }}">去注册</a></p>
{% endblock %}current_user 判断是否已登录。
login_user(user, remember=True) 登录用户，remember 开启 cookie 记住登录状态。
登出功能
添加登出路由：
@app.route('/logout')
@login_required
def logout():
logout_user()
flash('已登出！', 'success')
return redirect(url_for('home'))@login_required 限制只有登录用户才能访问。
保护文章发布
更新 /posts 路由，确保只有登录用户能发布文章：
@app.route('/posts', methods=['GET', 'POST'])
@login_required
def posts():
if request.method == 'POST':
title = request.form.get('title')
content = request.form.get('content')
post = Post(title=title, content=content, user_id=current_user.id)
db.session.add(post)
db.session.commit()
flash('文章发布成功！', 'success')
return redirect(url_for('posts'))
posts = Post.query.all()
return render_template('posts.html', posts=posts)添加 CSRF 保护
表单提交容易受到 CSRF 攻击。Flask-WTF 提供简单解决方案。安装：
pip install flask-wtf更新 app.py：
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, BooleanField, TextAreaField
from wtforms.validators import DataRequired, Email, Length
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[DataRequired(), Length(min=2, max=20)])
email = StringField('邮箱', validators=[DataRequired(), Email()])
password = PasswordField('密码', validators=[DataRequired()])
submit = SubmitField('注册')
class LoginForm(FlaskForm):
email = StringField('邮箱', validators=[DataRequired(), Email()])
password = PasswordField('密码', validators=[DataRequired()])
remember = BooleanField('记住我')
submit = SubmitField('登录')
@app.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit():
hashed_password = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
user = User(username=form.username.data, email=form.email.data, password=hashed_password)
try:
db.session.add(user)
db.session.commit()
flash('注册成功！请登录', 'success')
return redirect(url_for('login'))
except:
db.session.rollback()
flash('用户名或邮箱已存在！', 'danger')
return render_template('register.html', form=form)更新 templates/register.html：
{% extends 'base.html' %}
{% block title %}注册{% endblock %}
{% block content %}
<h1>注册新用户</h1>
{% if messages %}
{% for message in messages %}
<p class="{{ 'text-success' if message.category == 'success' else 'text-danger' }}">{{ message.message }}</p>
{% endfor %}
{% endif %}
<form method="POST">
{{ form.hidden_tag() }}
<div>
{{ form.username.label }} {{ form.username() }}
{% if form.username.errors %}
<p class="text-danger">{{ form.username.errors[0] }}</p>
{% endif %}
</div>
<div>
{{ form.email.label }} {{ form.email() }}
{% if form.email.errors %}
<p class="text-danger">{{ form.email.errors[0] }}</p>
{% endif %}
</div>
<div>
{{ form.password.label }} {{ form.password() }}
{% if form.password.errors %}
<p class="text-danger">{{ form.password.errors[0] }}</p>
{% endif %}
</div>
{{ form.submit() }}
</form>
<p>已有账号？<a href="{{ url_for('login') }}">去登录</a></p>
{% endblock %}form.hidden_tag() 生成 CSRF 令牌，validate_on_submit() 验证表单数据和 CSRF。登录表单类似改动。
小技巧与注意事项
密码安全：永远不要明文存储密码，用 Bcrypt 或 Argon2。
会话安全：生产环境中用强 SECRET_KEY，并启用 HTTPS。
表单验证：Flask-WTF 的验证器（如 Email()、Length()）能减少手动检查。
错误处理：捕获数据库异常，避免注册失败时崩溃。
章节 7：构建 RESTful API，解锁前后端分离
现代 Web 开发中，前后端分离是大势所趋。Flask 特别适合开发 RESTful API，代码简洁，响应速度快。这一章，我们将用 Flask 打造一个 RESTful API，支持文章的增删改查，还会处理认证和错误响应。
什么是 RESTful API？
REST（Representational State Transfer）是一种基于 HTTP 的架构风格，核心是资源（比如文章、用户）和标准操作（GET、POST、PUT、DELETE）。一个好的 RESTful API 应该：
资源导向：URL 表示资源，比如 /api/posts 表示文章列表。
HTTP 方法：GET 获取资源，POST 创建，PUT 更新，DELETE 删除。
状态码：用 HTTP 状态码表示操作结果，比如 200（成功）、404（未找到）、401（未授权）。
JSON 格式：数据通常以 JSON 传输，易于前端解析。
安装 Flask-RESTful
Flask-RESTful 是一个扩展，简化 API 开发。安装：
pip install flask-restful设置 API
我们为文章和用户创建 API，放在 app.py 中（稍后会重构代码结构）。更新 app.py：
from flask import Flask, render_template, request, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from flask_bcrypt import Bcrypt
from flask_restful import Api, Resource, reqparse
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, BooleanField, TextAreaField
from wtforms.validators import DataRequired, Email, Length
import os
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'your-secret-key'
db = SQLAlchemy(app)
bcrypt = Bcrypt(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'
api = Api(app)
# 模型（沿用之前）
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(20), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password = db.Column(db.String(60), nullable=False)
posts = db.relationship('Post', backref='author', lazy=True)
def __repr__(self):
return f"User('{self.username}', '{self.email}')"
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
def __repr__(self):
return f"Post('{self.title}')"
# 加载用户
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
# API 解析器
post_parser = reqparse.RequestParser()
post_parser.add_argument('title', type=str, required=True, help='标题不能为空')
post_parser.add_argument('content', type=str, required=True, help='内容不能为空')
# 文章 API
class PostAPI(Resource):
@login_required
def get(self, post_id=None):
if post_id:
post = Post.query.get_or_404(post_id)
return {'id': post.id, 'title': post.title, 'content': post.content, 'user_id': post.user_id}, 200
posts = Post.query.all()
return [{'id': p.id, 'title': p.title, 'content': p.content, 'user_id': p.user_id} for p in posts], 200
@login_required
def post(self):
args = post_parser.parse_args()
post = Post(title=args['title'], content=args['content'], user_id=current_user.id)
db.session.add(post)
db.session.commit()
return {'message': '文章创建成功', 'id': post.id}, 201
@login_required
def put(self, post_id):
post = Post.query.get_or_404(post_id)
if post.user_id != current_user.id:
return {'message': '无权修改此文章'}, 403
args = post_parser.parse_args()
post.title = args['title']
post.content = args['content']
db.session.commit()
return {'message': '文章更新成功'}, 200
@login_required
def delete(self, post_id):
post = Post.query.get_or_404(post_id)
if post.user_id != current_user.id:
return {'message': '无权删除此文章'}, 403
db.session.delete(post)
db.session.commit()
return {'message': '文章删除成功'}, 200
# 注册 API 路由
api.add_resource(PostAPI, '/api/posts', '/api/posts/<int:post_id>')
# 之前的路由（注册、登录、文章页面等，略）解释代码
Flask-RESTful：Api(app) 初始化 API，Resource 定义资源类，reqparse 验证请求参数。
PostAPI：
get：获取单篇文章（带 post_id）或所有文章。
post：创建新文章，需登录。
put：更新文章，仅限文章作者。
delete：删除文章，仅限作者。
状态码：201 表示创建成功，403 表示无权限，404 表示资源不存在。
@login_required：确保 API 需登录才能访问。
测试 API
运行应用，登录后用工具（如 Postman 或 curl）测试：
获取文章列表：GET http://127.0.0.1:5000/api/posts
curl -H "Cookie: session=你的会话ID" http://127.0.0.1:5000/api/posts返回 JSON 数组，列出所有文章。
创建文章：POST http://127.0.0.1:5000/api/posts
curl -X POST -H "Cookie: session=你的会话ID" -H "Content-Type: application/json" \ -d '{"title":"新文章","content":"这是内容"}' http://127.0.0.1:5000/api/posts返回 { "message": "文章创建成功", "id": 1 }。
更新文章：PUT http://127.0.0.1:5000/api/posts/1
删除文章：DELETE http://127.0.0.1:5000/api/posts/1
注意：会话 ID 需从登录后的 cookie 获取，或用 API 认证（后文会讲 JWT）。
API 认证（简单版）
目前我们用 Flask-Login 的会话认证，但 API 通常用 token。我们先实现一个简单的 token 认证，留到后续优化。安装 pyjwt：
pip install pyjwt添加 token 生成和验证：
import jwt
from datetime import datetime, timedelta
from functools import wraps
# Token 装饰器
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token:
return {'message': '缺少 token'}, 401
try:
data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
current_user = User.query.get(data['user_id'])
except:
return {'message': 'Token 无效'}, 401
return f(current_user, *args, **kwargs)
return decorated
# 生成 token
@app.route('/api/login', methods=['POST'])
def api_login():
email = request.json.get('email')
password = request.json.get('password')
user = User.query.filter_by(email=email).first()
if user and bcrypt.check_password_hash(user.password, password):
token = jwt.encode({
'user_id': user.id,
'exp': datetime.utcnow() + timedelta(hours=1)
}, app.config['SECRET_KEY'])
return {'token': token}, 200
return {'message': '登录失败'}, 401
# 修改 PostAPI 的 get 方法（示例）
class PostAPI(Resource):
@token_required
def get(self, current_user, post_id=None):
if post_id:
post = Post.query.get_or_404(post_id)
return {'id': post.id, 'title': post.title, 'content': post.content, 'user_id': post.user_id}, 200
posts = Post.query.all()
return [{'id': p.id, 'title': p.title, 'content': p.content, 'user_id': p.user_id} for p in posts], 200测试 token 登录：
curl -X POST -H "Content-Type: application/json" \
-d '{"email":"test@example.com","password":"yourpassword"}' http://127.0.0.1:5000/api/login返回 token 后，用它访问 API：
curl -H "Authorization: your_token_here" http://127.0.0.1:5000/api/posts小技巧与注意事项
JSON 格式：始终返回一致的 JSON 结构，比如 {'message': '', 'data': {}}。
错误处理：用 get_or_404 简化 404 处理。
安全性：生产环境用 HTTPS 传输 token，避免泄露。
扩展性：Flask-RESTful 支持字段过滤、分页等，适合复杂 API。
章节 8：部署上线，让全世界看到你的应用
开发完 Flask 应用，接下来是时候让它跑在云服务器上，接受来自全球的访问！这一章，我们将用 Gunicorn（应用服务器）和 Nginx（反向代理）部署到 Ubuntu 服务器，涵盖域名配置、HTTPS 证书和安全优化。
准备云服务器
我们以 Ubuntu 20.04（或 22.04）为例，推荐云服务商如 AWS EC2、DigitalOcean 或阿里云。步骤：
租用一台云服务器（1 核 2GB 内存够用）。
SSH 登录：ssh ubuntu@你的服务器IP。
更新系统：
sudo apt update && sudo apt upgrade -y
安装依赖
在服务器上安装 Python、pip 和虚拟环境：
sudo apt install python3 python3-pip python3-venv -y上传项目
将本地项目上传到服务器。推荐用 Git：
在本地项目目录初始化 Git：
git init git add . git commit -m "Initial commit"在服务器创建项目目录（如 /var/www/myapp）：
sudo mkdir -p /var/www/myapp sudo chown ubuntu:ubuntu /var/www/myapp本地推送代码到服务器（需先配置 SSH 或 Git 仓库），或用 scp：
scp -r ./my_flask_project ubuntu@你的服务器IP:/var/www/myapp
配置虚拟环境
在服务器的项目目录（/var/www/myapp）：
cd /var/www/myapp
python3 -m venv venv
source venv/bin/activate
pip install flask flask-sqlalchemy flask-login flask-bcrypt flask-restful flask-wtf pyjwt gunicorn测试 Gunicorn
Gunicorn 是一个生产级的 WSGI 服务器，适合运行 Flask 应用。测试：
gunicorn --workers 3 --bind 0.0.0.0:8000 app:app--workers 3：启动 3 个工作进程。
--bind 0.0.0.0:8000：监听 8000 端口。
app:app：指定模块 app.py 里的 Flask 实例 app。
访问 http://你的服务器IP:8000，应该看到应用运行。但别急，Gunicorn 不能直接暴露给公网，需用 Nginx 做反向代理。
配置 Nginx
安装 Nginx：
sudo apt install nginx -y创建 Nginx 配置文件 /etc/nginx/sites-available/myapp：
server {
listen 80;
server_name 你的域名 或 服务器IP;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
location /static {
alias /var/www/myapp/static;
}
}启用配置：
sudo ln -s /etc/nginx/sites-available/myapp /etc/nginx/sites-enabled/
sudo nginx -t # 测试配置
sudo systemctl restart nginx配置 systemd 服务
让 Gunicorn 随系统启动。创建 /etc/systemd/system/myapp.service：
[Unit]
Description=Gunicorn instance for myapp
After=network.target
[Service]
User=ubuntu
Group=www-data
WorkingDirectory=/var/www/myapp
Environment="PATH=/var/www/myapp/venv/bin"
ExecStart=/var/www/myapp/venv/bin/gunicorn --workers 3 --bind 0.0.0.0:8000 app:app
[Install]
WantedBy=multi-user.target启用服务：
sudo systemctl start myapp
sudo systemctl enable myapp配置 HTTPS
使用 Let’s Encrypt 获取免费 SSL 证书：
sudo apt install certbot python3-certbot-nginx -y
sudo certbot --nginx -d 你的域名按提示配置，Certbot 会自动更新 Nginx 配置，启用 HTTPS。
数据库迁移
本地 SQLite 数据库需迁移到服务器。如果用 MySQL，安装：
sudo apt install mysql-server -y创建数据库和用户：
sudo mysql
CREATE DATABASE myapp;
CREATE USER 'myappuser'@'localhost' IDENTIFIED BY 'yourpassword';
GRANT ALL PRIVILEGES ON myapp.* TO 'myappuser'@'localhost';
FLUSH PRIVILEGES;
EXIT;更新 app.py 的数据库 URI：
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://myappuser:yourpassword@localhost/myapp'使用 Flask-Migrate 管理表结构变更：
pip install flask-migrate初始化迁移：
from flask_migrate import Migrate
migrate = Migrate(app, db)在服务器运行：
flask db init
flask db migrate
flask db upgrade小技巧与注意事项
防火墙：用 ufw 开启 80 和 443 端口：
sudo ufw allow 80 sudo ufw allow 443 sudo ufw enable日志：检查 Gunicorn 日志：journalctl -u myapp.service。
性能：根据服务器性能调整 Gunicorn 的 --workers（通常 CPU 核心数 * 2 + 1）。
安全：禁用 Flask 调试模式（debug=False），定期更新依赖。
访问 https://你的域名，你的应用已经上线！是不是超激动？
章节 9：性能优化与扩展，让应用飞起来
当你的 Flask 应用上线后，用户量可能逐渐增加，性能瓶颈和代码维护问题就会暴露出来。这一章，我们将通过缓存、异步任务和 Blueprints 组织代码，让你的应用更高效、更易维护。
使用 Flask-Caching 加速响应
频繁查询数据库或计算密集型任务会拖慢响应速度。缓存是提升性能的神器！Flask-Caching 支持多种后端（如内存、Redis、Memcached），我们以 Redis 为例。
安装依赖
安装 Flask-Caching 和 Redis：
pip install flask-caching redis在服务器上安装 Redis：
sudo apt install redis-server -y
sudo systemctl enable redis配置缓存
更新 app.py：
from flask import Flask, render_template, jsonify
from flask_caching import Cache
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_HOST'] = 'localhost'
app.config['CACHE_REDIS_PORT'] = 6379
app.config['CACHE_DEFAULT_TIMEOUT'] = 300 # 缓存 5 分钟
db = SQLAlchemy(app)
cache = Cache(app)缓存视图
假设文章列表查询耗时较长，我们缓存它：
@app.route('/posts')
@cache.cached(timeout=60) # 缓存 1 分钟
def posts():
posts = Post.query.all()
return render_template('posts.html', posts=posts)访问 /posts，第一次查询数据库，后续请求直接从 Redis 读取缓存，速度飞快！
缓存 API
对 RESTful API 也可用缓存：
class PostAPI(Resource):
@cache.cached(timeout=60)
def get(self, post_id=None):
if post_id:
post = Post.query.get_or_404(post_id)
return {'id': post.id, 'title': post.title, 'content': post.content, 'user_id': post.user_id}, 200
posts = Post.query.all()
return [{'id': p.id, 'title': p.title, 'content': p.content, 'user_id': p.user_id} for p in posts], 200清除缓存
当文章更新或删除时，需清除缓存：
@cache.delete('view/posts')
def clear_posts_cache():
pass
@app.route('/posts', methods=['POST'])
@login_required
def create_post():
title = request.form.get('title')
content = request.form.get('content')
post = Post(title=title, content=content, user_id=current_user.id)
db.session.add(post)
db.session.commit()
clear_posts_cache() # 清除缓存
flash('文章发布成功！', 'success')
return redirect(url_for('posts'))小技巧
选择后端：小型项目用 simple（内存缓存），大型项目用 Redis 或 Memcached。
缓存键：用 @cache.memoize 缓存函数结果，适合动态参数。
超时时间：根据数据更新频率设置，比如静态内容可缓存更久。
异步任务与 Celery
有些任务（如发送邮件、生成报告）耗时长，阻塞用户请求会影响体验。Celery 是一个强大的异步任务队列，我们用它处理后台任务。
安装依赖
pip install celery redisRedis 作为消息队列（已安装）。
配置 Celery
在项目目录下创建 celery_config.py：
from celery import Celery
def make_celery(app):
celery = Celery(
app.import_name,
backend='redis://localhost:6379/0',
broker='redis://localhost:6379/0'
)
celery.conf.update(app.config)
return celery更新 app.py：
from celery_config import make_celery
app = Flask(__name__)
# 其他配置（略）
celery = make_celery(app)创建异步任务
假设我们想异步发送欢迎邮件。创建 tasks.py：
from celery import shared_task
import smtplib
from email.mime.text import MIMEText
@shared_task
def send_welcome_email(email, username):
msg = MIMEText(f'欢迎 {username} 加入我们的平台！')
msg['Subject'] = '欢迎注册'
msg['From'] = 'your_email@example.com'
msg['To'] = email
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('your_email@example.com', 'your_password')
server.send_message(msg)调用任务
在注册路由中触发：
@app.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit():
hashed_password = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
user = User(username=form.username.data, email=form.email.data, password=hashed_password)
try:
db.session.add(user)
db.session.commit()
send_welcome_email.delay(form.email.data, form.username.data) # 异步调用
flash('注册成功！请登录', 'success')
return redirect(url_for('login'))
except:
db.session.rollback()
flash('用户名或邮箱已存在！', 'danger')
return render_template('register.html', form=form)运行 Celery
在项目目录下，开启 Celery worker：
celery -A app.celery worker --loglevel=info另开终端运行 Redis 和 Flask 应用。注册新用户时，邮件会在后台发送，不阻塞用户请求！
小技巧
任务监控：用 Flower 监控 Celery 任务：pip install flower && celery -A app.celery flower。
生产环境：用 Supervisor 管理 Celery worker。
错误重试：为任务设置 retry 参数，处理临时失败。
使用 Blueprints 组织代码
项目变大后，app.py 会变得臃肿。Blueprints 是 Flask 提供的模块化工具，类似“迷你应用”。
重构项目结构
调整目录：
my_flask_project/
├── app/
│ ├── __init__.py
│ ├── main/
│ │ ├── __init__.py
│ │ ├── routes.py
│ ├── auth/
│ │ ├── __init__.py
│ │ ├── routes.py
│ │ ├── forms.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── routes.py
│ ├── models.py
│ ├── templates/
│ ├── static/
├── tasks.py
├── celery_config.py
├── config.py
└── run.py配置应用
创建 config.py：
class Config:
SQLALCHEMY_DATABASE_URI = 'sqlite:///site.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SECRET_KEY = 'your-secret-key'
CACHE_TYPE = 'redis'
CACHE_REDIS_HOST = 'localhost'
CACHE_REDIS_PORT = 6379
CACHE_DEFAULT_TIMEOUT = 300创建 app/__init__.py：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_bcrypt import Bcrypt
from flask_restful import Api
from flask_caching import Cache
from flask_migrate import Migrate
from config import Config
db = SQLAlchemy()
login_manager = LoginManager()
bcrypt = Bcrypt()
cache = Cache()
migrate = Migrate()
def create_app():
app = Flask(__name__)
app.config.from_object(Config)
db.init_app(app)
login_manager.init_app(app)
login_manager.login_view = 'auth.login'
bcrypt.init_app(app)
cache.init_app(app)
migrate.init_app(app, db)
api = Api(app)
from app.main.routes import main
from app.auth.routes import auth
from app.api.routes import api_bp
app.register_blueprint(main)
app.register_blueprint(auth, url_prefix='/auth')
app.register_blueprint(api_bp, url_prefix='/api')
return app主路由
创建 app/main/routes.py：
from flask import Blueprint, render_template
from app.models import Post
main = Blueprint('main', __name__)
@main.route('/')
def home():
return render_template('index.html', title='欢迎体验 Flask')
@main.route('/posts')
def posts():
posts = Post.query.all()
return render_template('posts.html', posts=posts)认证路由
创建 app/auth/forms.py：
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, BooleanField
from wtforms.validators import DataRequired, Email, Length
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[DataRequired(), Length(min=2, max=20)])
email = StringField('邮箱', validators=[DataRequired(), Email()])
password = PasswordField('密码', validators=[DataRequired()])
submit = SubmitField('注册')
class LoginForm(FlaskForm):
email = StringField('邮箱', validators=[DataRequired(), Email()])
password = PasswordField('密码', validators=[DataRequired()])
remember = BooleanField('记住我')
submit = SubmitField('登录')创建 app/auth/routes.py：
from flask import Blueprint, render_template, redirect, url_for, flash
from flask_login import login_user, logout_user, current_user, login_required
from app.models import User
from app.auth.forms import RegisterForm, LoginForm
from app import db, bcrypt
from tasks import send_welcome_email
auth = Blueprint('auth', __name__)
@auth.route('/register', methods=['GET', 'POST'])
def register():
if current_user.is_authenticated:
return redirect(url_for('main.home'))
form = RegisterForm()
if form.validate_on_submit():
hashed_password = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
user = User(username=form.username.data, email=form.email.data, password=hashed_password)
try:
db.session.add(user)
db.session.commit()
send_welcome_email.delay(form.email.data, form.username.data)
flash('注册成功！请登录', 'success')
return redirect(url_for('auth.login'))
except:
db.session.rollback()
flash('用户名或邮箱已存在！', 'danger')
return render_template('register.html', form=form)
@auth.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('main.home'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user and bcrypt.check_password_hash(user.password, form.password.data):
login_user(user, remember=form.remember.data)
flash('登录成功！', 'success')
return redirect(url_for('main.home'))
flash('邮箱或密码错误！', 'danger')
return render_template('login.html', form=form)
@auth.route('/logout')
@login_required
def logout():
logout_user()
flash('已登出！', 'success')
return redirect(url_for('main.home'))API 路由
创建 app/api/routes.py：
from flask import Blueprint, request
from flask_restful import Api, Resource, reqparse
from flask_login import login_required, current_user
from app.models import Post
from app import db
api_bp = Blueprint('api', __name__)
api = Api(api_bp)
post_parser = reqparse.RequestParser()
post_parser.add_argument('title', type=str, required=True, help='标题不能为空')
post_parser.add_argument('content', type=str, required=True, help='内容不能为空')
class PostAPI(Resource):
@login_required
def get(self, post_id=None):
if post_id:
post = Post.query.get_or_404(post_id)
return {'id': post.id, 'title': post.title, 'content': post.content, 'user_id': post.user_id}, 200
posts = Post.query.all()
return [{'id': p.id, 'title': p.title, 'content': p.content, 'user_id': p.user_id} for p in posts], 200
@login_required
def post(self):
args = post_parser.parse_args()
post = Post(title=args['title'], content=args['content'], user_id=current_user.id)
db.session.add(post)
db.session.commit()
return {'message': '文章创建成功', 'id': post.id}, 201
api.add_resource(PostAPI, '/posts', '/posts/<int:post_id>')运行应用
创建 run.py：
from app import create_app
app = create_app()
if __name__ == '__main__':
app.run(debug=True)运行：
python run.py访问 /、/auth/register、/api/posts，功能不变，但代码更清晰！
小技巧
URL 前缀：Blueprints 支持 url_prefix，适合模块化路由。
测试覆盖：重构后，测试每个 Blueprint 的路由。
扩展性：为新功能创建新 Blueprint，如 admin 或 blog。
章节 10：测试与调试技巧，打造高质量应用
没有测试的代码就像没有安全带的车，跑得快但随时可能翻车！这一章，我们用 unittest 编写测试用例，配置日志，介绍调试工具，确保你的 Flask 应用稳如老狗。
编写单元测试
Flask 自带测试客户端，结合 unittest 很好用。创建 tests/test_app.py：
import unittest
from app import create_app, db
from app.models import User, Post
class FlaskTestCase(unittest.TestCase):
def setUp(self):
self.app = create_app()
self.app.config['TESTING'] = True
self.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
self.client = self.app.test_client()
with self.app.app_context():
db.create_all()
def tearDown(self):
with self.app.app_context():
db.drop_all()
def test_home_page(self):
response = self.client.get('/')
self.assertEqual(response.status_code, 200)
self.assertIn(b'欢迎体验 Flask', response.data)
def test_register(self):
response = self.client.post('/auth/register', data={
'username': 'testuser',
'email': 'test@example.com',
'password': 'password123'
})
self.assertEqual(response.status_code, 302) # 重定向到登录
with self.app.app_context():
user = User.query.filter_by(email='test@example.com').first()
self.assertIsNotNone(user)
def test_login(self):
with self.app.app_context():
hashed_password = bcrypt.generate_password_hash('password123').decode('utf-8')
user = User(username='testuser', email='test@example.com', password=hashed_password)
db.session.add(user)
db.session.commit()
response = self.client.post('/auth/login', data={
'email': 'test@example.com',
'password': 'password123'
})
self.assertEqual(response.status_code, 302) # 重定向到首页
if __name__ == '__main__':
unittest.main()运行测试：
python -m unittest tests/test_app.py测试要点
setUp/tearDown：每次测试前创建内存数据库，测试后清理。
test_client：模拟 HTTP 请求，检查状态码和响应内容。
覆盖率：用 coverage 工具检查测试覆盖率：pip install coverage && coverage run -m unittest discover。
配置日志
日志是调试和监控的利器。配置 Python 的 logging 模块：
import logging
from logging.handlers import RotatingFileHandler
def setup_logging(app):
if not app.debug:
handler = RotatingFileHandler('logs/app.log', maxBytes=10000, backupCount=3)
handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
handler.setLevel(logging.INFO)
app.logger.addHandler(handler)
app.logger.setLevel(logging.INFO)
# 在 create_app 中调用
def create_app():
app = Flask(__name__)
app.config.from_object(Config)
setup_logging(app)
# 其他初始化（略）
return app日志示例
@app.route('/posts', methods=['POST'])
@login_required
def create_post():
app.logger.info(f'User {current_user.username} is creating a post')
try:
title = request.form.get('title')
content = request.form.get('content')
post = Post(title=title, content=content, user_id=current_user.id)
db.session.add(post)
db.session.commit()
app.logger.info(f'Post {title} created successfully')
flash('文章发布成功！', 'success')
return redirect(url_for('main.posts'))
except Exception as e:
app.logger.error(f'Error creating post: {str(e)}')
db.session.rollback()
flash('发布失败！', 'danger')
return redirect(url_for('main.posts'))日志保存在 logs/app.log，自动轮转，避免文件过大。
调试工具
Flask 调试模式：debug=True 显示交互式错误页面，但生产环境禁用。
Werkzeug 调试器：默认集成，点击错误页面可进入 Python 控制台。
pdb：在代码中加 import pdb; pdb.set_trace()，进入断点调试。
VS Code 调试：配置 launch.json：
{ "version": "0.2.0", "configurations": [ { "name": "Flask", "type": "python", "request": "launch", "module": "flask", "env": { "FLASK_APP": "run.py", "FLASK_ENV": "development" }, "args": ["run"], "jinja": true } ] }
小技巧与注意事项
测试环境：用 pytest 替代 unittest，支持更丰富的插件。
日志级别：生产环境用 INFO 或 WARNING，开发用 DEBUG。
性能监控：用 New Relic 或 Sentry 监控线上应用。