Flask¶
从基础到高级，涵盖路由、请求响应、模板、ORM、认证、蓝图、异步、部署等核心知识点。以 Flask 3.x + Flask-SQLAlchemy 3.x 为基准。
目录¶
一、基础篇¶
1.1 安装与项目结构¶
# 安装
pip install flask flask-sqlalchemy flask-migrate flask-login flask-jwt-extended
pip install flask-wtf flask-caching celery redis
# 推荐项目结构（中大型项目）
myapp/
├── app/
│ ├── __init__.py # 应用工厂
│ ├── models/ # 数据模型
│ │ ├── __init__.py
│ │ └── user.py
│ ├── api/ # 蓝图（API）
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── user.py
│ ├── templates/ # Jinja2 模板
│ ├── static/ # 静态文件
│ └── utils/ # 工具函数
├── migrations/ # 数据库迁移文件
├── tests/ # 测试
├── config.py # 配置
├── requirements.txt
└── run.py # 启动入口
1.2 路由与视图¶
from flask import Flask
app = Flask(__name__)
# 基本路由
@app.route('/')
def index():
return 'Hello Flask!'
# 路由变量
@app.route('/user/<int:user_id>')
def get_user(user_id):
return f'User {user_id}'
@app.route('/path/<path:subpath>') # path 类型可包含斜杠
def show_path(subpath):
return f'Path: {subpath}'
# 指定 HTTP 方法
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
return 'do login'
return 'show login form'
# 路由变量类型
# <int:id> 整数
# <float:num> 浮点数
# <string:name> 字符串（默认，不含/）
# <path:p> 路径（含/）
# <uuid:uid> UUID
# URL 构建
from flask import url_for
with app.test_request_context():
print(url_for('get_user', user_id=1)) # /user/1
print(url_for('index', _external=True)) # http://localhost/
1.3 请求与响应¶
from flask import request, jsonify, make_response, redirect, abort
# 请求对象
request.method # GET / POST / PUT ...
request.args # URL 查询参数（?key=val）
request.form # 表单数据（POST body）
request.json # JSON body（自动解析）
request.data # 原始 body（bytes）
request.files # 上传文件
request.headers # 请求头
request.cookies # Cookie
request.remote_addr # 客户端 IP
request.url # 完整 URL
request.path # 路径部分
request.get_json() # 手动解析 JSON（force=True 忽略 Content-Type）
# 响应方式
return 'plain text' # 纯文本
return '<h1>HTML</h1>' # HTML
return jsonify({'code': 0, 'data': {}}) # JSON（推荐）
return jsonify(data), 201 # 带状态码
return redirect(url_for('index')) # 重定向
abort(404) # 抛出 HTTP 错误
# 自定义响应
response = make_response(jsonify({'msg': 'ok'}), 200)
response.headers['X-Custom-Header'] = 'value'
response.set_cookie('token', 'xxx', httponly=True, max_age=3600)
return response
# 文件下载
from flask import send_file, send_from_directory
return send_file('/path/to/file.pdf', as_attachment=True)
return send_from_directory('uploads', filename)
1.4 配置管理¶
# config.py
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')
SQLALCHEMY_TRACK_MODIFICATIONS = False
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'
class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig
}
# 加载配置
app.config.from_object('config.DevelopmentConfig')
app.config.from_envvar('APP_CONFIG_FILE') # 从环境变量指定的文件加载
app.config.from_mapping(DEBUG=True) # 直接设置
# 读取配置
app.config['SECRET_KEY']
app.config.get('CUSTOM_KEY', 'default')
1.5 上下文¶
# 应用上下文（Application Context）
# 在请求或 CLI 命令之外使用时需手动推送
with app.app_context():
db.create_all()
# 请求上下文（Request Context）
# 请求期间自动激活，可访问 request、g、session
# g 对象：请求内的临时存储
from flask import g
@app.before_request
def load_user():
g.current_user = get_user_from_token()
@app.route('/profile')
def profile():
return jsonify(g.current_user.to_dict())
# current_app：在无法直接访问 app 的地方使用
from flask import current_app
current_app.config['SECRET_KEY']
current_app.logger.info('log message')
二、模板篇¶
2.1 Jinja2 语法¶
2.2 模板继承¶
三、ORM 篇（Flask-SQLAlchemy）¶
3.1 安装与配置¶
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
db.init_app(app)
return app
# config.py 数据库连接串
# SQLite
SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'
# MySQL
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:pass@localhost:3306/mydb'
# PostgreSQL
SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://user:pass@localhost:5432/mydb'
# 连接池配置
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'max_overflow': 20,
'pool_pre_ping': True, # 自动检测断开的连接
'pool_recycle': 3600 # 连接1小时后回收
}
3.2 模型定义¶
# app/models/user.py
from app import db
from datetime import datetime
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
username = db.Column(db.String(50), nullable=False, unique=True)
email = db.Column(db.String(120), nullable=False, unique=True, index=True)
password = db.Column(db.String(256), nullable=False)
age = db.Column(db.Integer)
balance = db.Column(db.Numeric(12, 2), default=0.00)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f'<User {self.username}>'
def to_dict(self):
return {
'id': self.id,
'username': self.username,
'email': self.email,
'is_active': self.is_active,
'created_at': self.created_at.isoformat()
}
3.3 字段类型与约束¶
常用字段类型
类型 |
说明 |
|---|---|
|
整数 |
|
大整数 |
|
浮点数 |
|
精确小数（推荐金额） |
|
定长字符串 VARCHAR |
|
长文本 TEXT |
|
布尔值 |
|
日期时间 |
|
日期 |
|
时间 |
|
JSON（MySQL 5.7+ / PG） |
|
二进制 BLOB |
|
枚举 |
字段约束参数
db.Column(
db.String(50),
primary_key=True, # 主键
nullable=False, # 非空
unique=True, # 唯一
index=True, # 创建索引
default='value', # Python 默认值
server_default='0', # 数据库默认值
comment='字段说明' # 字段注释
)
联合索引 / 表级约束
class Order(db.Model):
__tablename__ = 'orders'
__table_args__ = (
db.UniqueConstraint('user_id', 'order_no', name='uq_user_order'),
db.Index('idx_user_status', 'user_id', 'status'),
db.CheckConstraint('amount > 0', name='ck_amount_positive'),
{'mysql_engine': 'InnoDB', 'mysql_charset': 'utf8mb4'}
)
id = db.Column(db.BigInteger, primary_key=True)
user_id = db.Column(db.BigInteger, nullable=False)
order_no = db.Column(db.String(32), nullable=False)
status = db.Column(db.String(20), default='pending')
amount = db.Column(db.Numeric(12, 2))
3.4 关系映射¶
# 一对多：User → Order
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.BigInteger, primary_key=True)
orders = db.relationship('Order', backref='user', lazy='dynamic')
# lazy='select' 默认，访问时查询
# lazy='dynamic' 返回查询对象，可继续过滤
# lazy='joined' JOIN 一次性加载
# lazy='subquery' 子查询加载
class Order(db.Model):
__tablename__ = 'orders'
id = db.Column(db.BigInteger, primary_key=True)
user_id = db.Column(db.BigInteger, db.ForeignKey('users.id', ondelete='CASCADE'))
amount = db.Column(db.Numeric(12, 2))
# 使用
user = User.query.get(1)
user.orders.all() # dynamic lazy 下需调用 .all()
user.orders.filter_by(status='paid').all()
order = Order.query.get(1)
order.user # 通过 backref 访问关联用户
# 多对多：User ↔ Role（需中间表）
user_roles = db.Table('user_roles',
db.Column('user_id', db.BigInteger, db.ForeignKey('users.id')),
db.Column('role_id', db.BigInteger, db.ForeignKey('roles.id'))
)
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.BigInteger, primary_key=True)
roles = db.relationship('Role', secondary=user_roles, backref='users')
class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.BigInteger, primary_key=True)
name = db.Column(db.String(50))
# 使用
user.roles.append(role)
user.roles.remove(role)
role.users.all() # backref 反向访问
# 一对一：User → Profile
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.BigInteger, primary_key=True)
profile = db.relationship('Profile', backref='user', uselist=False) # uselist=False 一对一
class Profile(db.Model):
__tablename__ = 'profiles'
id = db.Column(db.BigInteger, primary_key=True)
user_id = db.Column(db.BigInteger, db.ForeignKey('users.id'), unique=True)
bio = db.Column(db.Text)
3.5 增删改查¶
# ---- 增 ----
user = User(username='alice', email='alice@example.com', password='hashed')
db.session.add(user)
db.session.commit()
# 批量插入
db.session.bulk_insert_mappings(User, [
{'username': 'bob', 'email': 'bob@example.com'},
{'username': 'carol', 'email': 'carol@example.com'},
])
db.session.commit()
# ---- 查 ----
# 主键查询
user = db.session.get(User, 1) # SQLAlchemy 2.x 推荐
user = User.query.get(1) # 旧写法（将废弃）
user = User.query.get_or_404(1) # 不存在则返回 404
# 条件查询
users = User.query.filter_by(is_active=True).all()
users = User.query.filter(User.age > 18).all()
users = User.query.filter(
User.is_active == True,
User.age >= 18
).all()
# 常用查询方法
User.query.all() # 全部
User.query.first() # 第一条
User.query.count() # 数量
User.query.one() # 精确一条（多条或零条抛异常）
User.query.one_or_none() # 零条返回None，多条抛异常
# 排序
User.query.order_by(User.created_at.desc()).all()
User.query.order_by(User.age.asc(), User.id.desc()).all()
# 分页
page = User.query.order_by(User.id).paginate(page=1, per_page=20, error_out=False)
page.items # 当前页数据
page.total # 总数量
page.pages # 总页数
page.has_next # 是否有下一页
page.next_num # 下一页页码
# 字段过滤
users = User.query.with_entities(User.id, User.username).all()
# 模糊查询
User.query.filter(User.username.like('%alice%')).all()
User.query.filter(User.username.ilike('%alice%')).all() # 不区分大小写
# IN 查询
User.query.filter(User.id.in_([1, 2, 3])).all()
User.query.filter(~User.id.in_([1, 2, 3])).all() # NOT IN
# NULL 查询
User.query.filter(User.deleted_at.is_(None)).all()
User.query.filter(User.deleted_at.isnot(None)).all()
# OR 查询
from sqlalchemy import or_, and_
User.query.filter(or_(User.age < 18, User.age > 60)).all()
# JOIN
db.session.query(User, Order).join(Order, User.id == Order.user_id).all()
User.query.join(Order).filter(Order.status == 'paid').all()
# 聚合
from sqlalchemy import func
db.session.query(func.count(User.id)).scalar()
db.session.query(func.sum(Order.amount)).filter(Order.user_id == 1).scalar()
db.session.query(Order.user_id, func.count(Order.id).label('cnt'))\
.group_by(Order.user_id).all()
# ---- 改 ----
user = db.session.get(User, 1)
user.username = 'new_name'
db.session.commit()
# 批量更新（高效）
User.query.filter(User.is_active == False).update({'deleted': True})
db.session.commit()
# ---- 删 ----
user = db.session.get(User, 1)
db.session.delete(user)
db.session.commit()
# 批量删除
User.query.filter(User.created_at < cutoff_date).delete()
db.session.commit()
# ---- 事务 ----
try:
user = User(username='alice')
db.session.add(user)
order = Order(user=user, amount=99.9)
db.session.add(order)
db.session.commit()
except Exception as e:
db.session.rollback()
raise e
3.6 迁移（Flask-Migrate）¶
# 安装
pip install flask-migrate
# app/__init__.py
from flask_migrate import Migrate
migrate = Migrate()
def create_app():
app = Flask(__name__)
db.init_app(app)
migrate.init_app(app, db)
return app
# 初始化迁移目录（只执行一次）
flask db init
# 生成迁移文件（检测模型变化）
flask db migrate -m "add users table"
# 执行迁移（应用到数据库）
flask db upgrade
# 回滚迁移
flask db downgrade
# 查看迁移历史
flask db history
# 查看当前版本
flask db current
四、表单与验证篇¶
4.1 Flask-WTF¶
# pip install flask-wtf
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, IntegerField, SelectField, BooleanField
from wtforms.validators import DataRequired, Email, Length, EqualTo, NumberRange, Optional
class RegisterForm(FlaskForm):
username = StringField('用户名', validators=[
DataRequired(message='用户名不能为空'),
Length(min=3, max=50, message='用户名长度3-50位')
])
email = StringField('邮箱', validators=[
DataRequired(),
Email(message='邮箱格式不正确')
])
password = PasswordField('密码', validators=[
DataRequired(),
Length(min=6, message='密码至少6位')
])
confirm = PasswordField('确认密码', validators=[
EqualTo('password', message='两次密码不一致')
])
age = IntegerField('年龄', validators=[
Optional(),
NumberRange(min=1, max=150)
])
# 视图中使用
@app.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit(): # POST 且验证通过
user = User(
username=form.username.data,
email=form.email.data,
)
db.session.add(user)
db.session.commit()
return redirect(url_for('login'))
return render_template('register.html', form=form)
4.2 API 数据验证（marshmallow）¶
# pip install marshmallow flask-marshmallow
from marshmallow import Schema, fields, validate, validates, ValidationError
class UserSchema(Schema):
id = fields.Int(dump_only=True) # 只序列化，不反序列化
username = fields.Str(required=True, validate=validate.Length(min=3, max=50))
email = fields.Email(required=True)
age = fields.Int(validate=validate.Range(min=1, max=150), load_default=None)
@validates('username')
def validate_username(self, value):
if User.query.filter_by(username=value).first():
raise ValidationError('用户名已存在')
user_schema = UserSchema()
users_schema = UserSchema(many=True)
# 序列化（对象 → dict）
result = user_schema.dump(user)
# 反序列化（dict → 验证后的 dict）
try:
data = user_schema.load(request.json)
except ValidationError as e:
return jsonify({'errors': e.messages}), 422
五、认证与权限篇¶
5.1 Session¶
from flask import session
# 设置 session（需要 SECRET_KEY）
session['user_id'] = user.id
session.permanent = True # 持久化 session
app.permanent_session_lifetime = timedelta(days=7)
# 读取
user_id = session.get('user_id')
# 删除
session.pop('user_id', None)
session.clear() # 清空所有
5.2 Flask-Login¶
# pip install flask-login
from flask_login import LoginManager, UserMixin, login_user, logout_user, \
login_required, current_user
login_manager = LoginManager()
login_manager.login_view = 'auth.login' # 未登录时跳转的端点
class User(UserMixin, db.Model): # 必须继承 UserMixin
# UserMixin 提供 is_authenticated, is_active, get_id() 等
@property
def is_active(self):
return self._is_active # 可自定义
@login_manager.user_loader
def load_user(user_id):
return db.session.get(User, int(user_id)) # 根据 ID 加载用户
# 登录
@app.route('/login', methods=['POST'])
def login():
user = User.query.filter_by(email=request.json['email']).first()
if user and check_password_hash(user.password, request.json['password']):
login_user(user, remember=True)
return jsonify({'msg': 'ok'})
return jsonify({'msg': '账号或密码错误'}), 401
# 注销
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('index'))
# 保护路由
@app.route('/profile')
@login_required
def profile():
return jsonify(current_user.to_dict())
5.3 JWT（Flask-JWT-Extended）¶
# pip install flask-jwt-extended
from flask_jwt_extended import (
JWTManager, create_access_token, create_refresh_token,
jwt_required, get_jwt_identity, get_jwt
)
app.config['JWT_SECRET_KEY'] = 'jwt-secret'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=2)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt = JWTManager(app)
# 登录颁发 token
@app.route('/login', methods=['POST'])
def login():
user = User.query.filter_by(email=request.json['email']).first()
if not user or not user.check_password(request.json['password']):
return jsonify({'msg': '认证失败'}), 401
access_token = create_access_token(identity=user.id, additional_claims={'role': user.role})
refresh_token = create_refresh_token(identity=user.id)
return jsonify(access_token=access_token, refresh_token=refresh_token)
# 刷新 token
@app.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
user_id = get_jwt_identity()
access_token = create_access_token(identity=user_id)
return jsonify(access_token=access_token)
# 保护路由
@app.route('/profile')
@jwt_required()
def profile():
user_id = get_jwt_identity() # 从 token 中获取用户ID
claims = get_jwt() # 获取全部声明
user = db.session.get(User, user_id)
return jsonify(user.to_dict())
# Token 黑名单（注销）
from flask_jwt_extended import get_jwt
blocklist = set()
@jwt.token_in_blocklist_loader
def check_if_token_revoked(jwt_header, jwt_payload):
return jwt_payload['jti'] in blocklist
@app.route('/logout', methods=['DELETE'])
@jwt_required()
def logout():
blocklist.add(get_jwt()['jti'])
return jsonify({'msg': '已注销'})
5.4 权限控制¶
from functools import wraps
from flask_jwt_extended import get_jwt
# 自定义权限装饰器
def require_role(*roles):
def decorator(f):
@wraps(f)
@jwt_required()
def decorated(*args, **kwargs):
claims = get_jwt()
if claims.get('role') not in roles:
return jsonify({'msg': '权限不足'}), 403
return f(*args, **kwargs)
return decorated
return decorator
# 使用
@app.route('/admin/users')
@require_role('admin', 'superadmin')
def admin_users():
return jsonify(users_schema.dump(User.query.all()))
六、进阶篇¶
6.1 蓝图（Blueprint）¶
蓝图将应用拆分为多个模块，各自管理路由、模板、静态文件。
# app/api/auth.py
from flask import Blueprint
auth_bp = Blueprint('auth', __name__, url_prefix='/api/auth')
@auth_bp.route('/login', methods=['POST'])
def login():
...
@auth_bp.route('/register', methods=['POST'])
def register():
...
# app/__init__.py
def create_app():
app = Flask(__name__)
from app.api.auth import auth_bp
from app.api.user import user_bp
from app.api.order import order_bp
app.register_blueprint(auth_bp)
app.register_blueprint(user_bp)
app.register_blueprint(order_bp)
return app
6.2 应用工厂模式¶
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
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.api.auth import auth_bp
from app.api.user import user_bp
app.register_blueprint(auth_bp)
app.register_blueprint(user_bp)
return app
# run.py
from app import create_app
app = create_app('development')
if __name__ == '__main__':
app.run()
6.3 钩子函数¶
# 全局钩子
@app.before_request
def before_request():
g.start_time = time.time() # 每个请求前执行
@app.after_request
def after_request(response):
duration = time.time() - g.start_time
response.headers['X-Response-Time'] = f'{duration:.3f}s'
return response # 必须返回 response
@app.teardown_request
def teardown(exception):
db.session.remove() # 请求结束后执行（无论成功失败）
@app.teardown_appcontext
def teardown_appcontext(exception):
pass # 应用上下文销毁时执行
# 蓝图钩子（只作用于该蓝图）
@auth_bp.before_request
def check_auth():
pass
6.4 错误处理¶
from flask import jsonify
# 全局错误处理
@app.errorhandler(404)
def not_found(e):
return jsonify({'code': 404, 'msg': '资源不存在'}), 404
@app.errorhandler(500)
def server_error(e):
return jsonify({'code': 500, 'msg': '服务器内部错误'}), 500
@app.errorhandler(403)
def forbidden(e):
return jsonify({'code': 403, 'msg': '无权限'}), 403
# 自定义业务异常
class AppError(Exception):
def __init__(self, message, code=400):
self.message = message
self.code = code
@app.errorhandler(AppError)
def handle_app_error(e):
return jsonify({'code': e.code, 'msg': e.message}), e.code
# 抛出
raise AppError('用户名已存在', code=409)
6.5 中间件¶
# WSGI 中间件
class RequestLoggingMiddleware:
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
# 请求前
print(f"Request: {environ['REQUEST_METHOD']} {environ['PATH_INFO']}")
return self.app(environ, start_response)
app.wsgi_app = RequestLoggingMiddleware(app.wsgi_app)
# 用 before_request / after_request 更简单（推荐）
6.6 信号（Signal）¶
from blinker import Namespace
# 自定义信号
_signals = Namespace()
user_registered = _signals.signal('user-registered')
# 发送信号
user_registered.send(app, user=new_user)
# 订阅信号
@user_registered.connect_via(app)
def on_user_registered(sender, user, **kwargs):
send_welcome_email(user.email)
七、REST API 篇¶
7.1 RESTful 设计规范¶
GET /api/users # 获取用户列表
POST /api/users # 创建用户
GET /api/users/<id> # 获取单个用户
PUT /api/users/<id> # 全量更新
PATCH /api/users/<id> # 部分更新
DELETE /api/users/<id> # 删除用户
7.2 统一响应格式¶
# utils/response.py
from flask import jsonify
def success(data=None, msg='success', code=200):
return jsonify({'code': 0, 'msg': msg, 'data': data}), code
def error(msg='error', code=400, errors=None):
resp = {'code': code, 'msg': msg}
if errors:
resp['errors'] = errors
return jsonify(resp), code
# 使用
return success(data=user.to_dict(), code=201)
return error(msg='用户不存在', code=404)
7.3 典型 CRUD 视图¶
# app/api/user.py
from flask import Blueprint, request
from app import db
from app.models.user import User
from app.utils.response import success, error
user_bp = Blueprint('user', __name__, url_prefix='/api/users')
@user_bp.route('/', methods=['GET'])
@jwt_required()
def list_users():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
keyword = request.args.get('keyword', '')
query = User.query.filter(User.is_active == True)
if keyword:
query = query.filter(User.username.like(f'%{keyword}%'))
pagination = query.order_by(User.id.desc()).paginate(page=page, per_page=per_page)
return success({
'list': [u.to_dict() for u in pagination.items],
'total': pagination.total,
'page': page,
'pages': pagination.pages
})
@user_bp.route('/<int:user_id>', methods=['GET'])
@jwt_required()
def get_user(user_id):
user = db.session.get(User, user_id) or abort(404)
return success(user.to_dict())
@user_bp.route('/', methods=['POST'])
def create_user():
data = request.get_json()
# 验证...
user = User(**data)
db.session.add(user)
db.session.commit()
return success(user.to_dict(), code=201)
@user_bp.route('/<int:user_id>', methods=['PUT'])
@jwt_required()
def update_user(user_id):
user = db.session.get(User, user_id) or abort(404)
data = request.get_json()
for key, value in data.items():
setattr(user, key, value)
db.session.commit()
return success(user.to_dict())
@user_bp.route('/<int:user_id>', methods=['DELETE'])
@jwt_required()
def delete_user(user_id):
user = db.session.get(User, user_id) or abort(404)
db.session.delete(user)
db.session.commit()
return success(msg='删除成功')
八、缓存与任务篇¶
8.1 Flask-Caching¶
# pip install flask-caching
from flask_caching import Cache
cache = Cache()
# config.py
CACHE_TYPE = 'RedisCache'
CACHE_REDIS_URL = 'redis://localhost:6379/1'
CACHE_DEFAULT_TIMEOUT = 300
# app/__init__.py
cache.init_app(app)
# 使用装饰器缓存视图
@app.route('/api/hot-articles')
@cache.cached(timeout=60, key_prefix='hot_articles')
def hot_articles():
return jsonify(Article.query.order_by(Article.views.desc()).limit(10).all())
# 缓存函数结果（带参数）
@cache.memoize(timeout=300)
def get_user_info(user_id):
return db.session.get(User, user_id).to_dict()
# 主动删除缓存
cache.delete('hot_articles')
cache.delete_memoized(get_user_info, user_id)
cache.clear() # 清除所有
# 手动操作缓存
cache.set('key', value, timeout=60)
cache.get('key')
cache.delete('key')
8.2 Celery 异步任务¶
# pip install celery redis
# celery_app.py
from celery import Celery
def make_celery(app):
celery = Celery(app.import_name)
celery.conf.update(
broker_url='redis://localhost:6379/0',
result_backend='redis://localhost:6379/0',
task_serializer='json',
result_serializer='json',
timezone='Asia/Shanghai'
)
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
# tasks.py
from celery_app import celery
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, to, subject, body):
try:
# 发送邮件逻辑
email_service.send(to, subject, body)
except Exception as exc:
raise self.retry(exc=exc)
@celery.task
def generate_report(user_id):
# 耗时操作
pass
# 定时任务
from celery.schedules import crontab
celery.conf.beat_schedule = {
'daily-report': {
'task': 'tasks.generate_report',
'schedule': crontab(hour=8, minute=0), # 每天早8点
'args': (1,)
}
}
# 调用异步任务
send_email.delay('user@example.com', '欢迎', '欢迎注册!')
send_email.apply_async(args=['user@example.com', '欢迎', '...'], countdown=60) # 60秒后执行
# 启动 worker
# celery -A tasks worker --loglevel=info
# celery -A tasks beat --loglevel=info # 定时任务调度
九、测试篇¶
9.1 单元测试¶
# tests/conftest.py
import pytest
from app import create_app, db
@pytest.fixture
def app():
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.session.remove()
db.drop_all()
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def runner(app):
return app.test_cli_runner()
# tests/test_user.py
import json
def test_create_user(client):
response = client.post('/api/users/', json={
'username': 'testuser',
'email': 'test@example.com',
'password': '123456'
})
assert response.status_code == 201
data = response.get_json()
assert data['data']['username'] == 'testuser'
def test_get_user_not_found(client):
response = client.get('/api/users/9999')
assert response.status_code == 404
def test_login(client):
# 先注册
client.post('/api/users/', json={'username': 'u', 'email': 'u@a.com', 'password': '123456'})
# 登录
response = client.post('/api/auth/login', json={'email': 'u@a.com', 'password': '123456'})
assert response.status_code == 200
assert 'access_token' in response.get_json()
9.2 Mock 外部依赖¶
from unittest.mock import patch, MagicMock
def test_send_email(client):
with patch('app.services.email.send') as mock_send:
mock_send.return_value = True
response = client.post('/api/register', json={...})
assert mock_send.called
assert response.status_code == 201
十、部署篇¶
10.1 Gunicorn¶
# pip install gunicorn
# 基本启动
gunicorn -w 4 -b 0.0.0.0:8000 "app:create_app()"
# 常用参数
gunicorn \
--workers 4 \ # worker 进程数（推荐：CPU核数 * 2 + 1）
--worker-class gevent \ # 异步 worker（需 pip install gevent）
--bind 0.0.0.0:8000 \
--timeout 120 \ # 请求超时
--max-requests 1000 \ # worker 处理N个请求后重启（防内存泄露）
--max-requests-jitter 100 \
--access-logfile /var/log/gunicorn/access.log \
--error-logfile /var/log/gunicorn/error.log \
"app:create_app()"
10.2 Nginx 配置¶
upstream flask_app {
server 127.0.0.1:8000;
# 多实例
# server 127.0.0.1:8001;
# server 127.0.0.1:8002;
}
server {
listen 80;
server_name yourdomain.com;
# 静态文件直接由 Nginx 处理
location /static/ {
alias /path/to/app/static/;
expires 30d;
add_header Cache-Control "public, immutable";
}
# 上传文件
location /uploads/ {
alias /path/to/uploads/;
}
location / {
proxy_pass http://flask_app;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 120;
client_max_body_size 20M;
}
}
10.3 Docker 部署¶
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]
# docker-compose.yml
version: "3.8"
services:
web:
build: .
ports:
- "8000:8000"
environment:
- FLASK_ENV=production
- DATABASE_URL=mysql+pymysql://user:pass@db:3306/mydb
- REDIS_URL=redis://redis:6379/0
depends_on:
- db
- redis
restart: unless-stopped
db:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: rootpass
MYSQL_DATABASE: mydb
MYSQL_USER: user
MYSQL_PASSWORD: pass
volumes:
- mysql_data:/var/lib/mysql
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/conf.d/default.conf
depends_on:
- web
volumes:
mysql_data:
redis_data:
10.4 环境变量管理¶
# .env（不提交 git）
SECRET_KEY=your-secret-key
DATABASE_URL=mysql+pymysql://user:pass@localhost:3306/mydb
REDIS_URL=redis://localhost:6379/0
JWT_SECRET_KEY=your-jwt-secret
# pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()
# 或在 config.py 中直接用 os.environ
import os
SECRET_KEY = os.environ.get('SECRET_KEY')
DATABASE_URL = os.environ.get('DATABASE_URL')
常用扩展汇总¶
扩展 |
功能 |
|---|---|
Flask-SQLAlchemy |
ORM |
Flask-Migrate |
数据库迁移 |
Flask-Login |
Session 认证 |
Flask-JWT-Extended |
JWT 认证 |
Flask-WTF |
表单验证 + CSRF |
Flask-Marshmallow |
序列化/反序列化/验证 |
Flask-Caching |
缓存（支持 Redis） |
Flask-CORS |
跨域处理 |
Flask-Mail |
邮件发送 |
Flask-Limiter |
接口限流 |
Flask-SocketIO |
WebSocket |
Celery |
异步任务队列 |
python-dotenv |
环境变量管理 |
参考资源¶
Flask 官方文档：https://flask.palletsprojects.com/
Flask-SQLAlchemy：https://flask-sqlalchemy.palletsprojects.com/
Flask-Migrate：https://flask-migrate.readthedocs.io/
Flask-JWT-Extended：https://flask-jwt-extended.readthedocs.io/
SQLAlchemy 文档：https://docs.sqlalchemy.org/