跳到主要内容

Flask 速查表

快速开始

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
return 'Hello, Flask!'

if __name__ == '__main__':
app.run(debug=True)

路由

# 基本路由
@app.route('/path')
def view():
pass

# 动态路由
@app.route('/user/<username>')
def user(username):
pass

# 类型转换器
@app.route('/post/<int:post_id>') # 整数
@app.route('/price/<float:price>') # 浮点数
@app.route('/path/<path:subpath>') # 路径(含斜杠)
@app.route('/uuid/<uuid:id>') # UUID

# HTTP 方法
@app.route('/api', methods=['GET', 'POST', 'PUT', 'DELETE'])
def api():
pass

# 专用装饰器(推荐)
@app.get('/users')
def get_users(): pass

@app.post('/users')
def create_user(): pass

# URL 构建
from flask import url_for
url_for('view_name', arg='value')
url_for('static', filename='style.css')

请求对象

from flask import request

# 表单数据(POST)
request.form['key'] # 获取值,不存在则报错
request.form.get('key', default) # 获取值,不存在返回默认值

# URL 参数(GET)
request.args.get('page', 1, type=int)

# JSON 数据
request.get_json() # 解析 JSON
request.json # 直接访问 JSON 数据

# 文件上传
file = request.files['file']
filename = file.filename
file.save(f'/path/{filename}')

# 请求头
request.headers.get('Content-Type')
request.headers.get('Authorization')

# 其他属性
request.method # HTTP 方法:'GET', 'POST' 等
request.url # 完整 URL
request.path # 路径部分
request.endpoint # 端点名称
request.cookies # Cookie 字典
request.remote_addr # 客户端 IP
request.user_agent # 用户代理

响应

from flask import make_response, jsonify, redirect, abort

# 字符串响应
return 'Hello'

# 元组响应 (内容, 状态码, 头)
return 'Not Found', 404
return 'Created', 201, {'Location': url}

# JSON 响应
return jsonify({'key': 'value'})
return jsonify(data=[1, 2, 3], status='ok'), 200

# 自定义响应对象
response = make_response('Content')
response.status_code = 201
response.headers['X-Custom'] = 'Value'
response.set_cookie('name', 'value', max_age=3600)
return response

# 重定向
return redirect(url_for('index'))
return redirect('/path')
return redirect(url_for('login'), code=301)

# 中止请求
abort(404) # 404 Not Found
abort(403) # 403 Forbidden
abort(400, 'Bad Request')

模板 (Jinja2)

Python 端

from flask import render_template

# 渲染模板
return render_template('template.html', var=value, items=items)

模板语法

<!-- 变量输出 -->
{{ variable }}
{{ user.name }}
{{ items[0] }}

<!-- 过滤器 -->
{{ name|upper }} <!-- 大写 -->
{{ text|truncate(50) }} <!-- 截断 -->
{{ content|safe }} <!-- 不转义 HTML -->
{{ price|round(2) }} <!-- 四舍五入 -->
{{ items|length }} <!-- 长度 -->
{{ text|default('N/A') }} <!-- 默认值 -->

<!-- 条件判断 -->
{% if user.is_admin %}
<p>管理员</p>
{% elif user.is_active %}
<p>活跃用户</p>
{% else %}
<p>普通用户</p>
{% endif %}

<!-- 循环 -->
{% for item in items %}
{{ loop.index }} <!-- 索引(从1开始) -->
{{ loop.index0 }} <!-- 索引(从0开始) -->
{{ loop.first }} <!-- 是否第一个 -->
{{ loop.last }} <!-- 是否最后一个 -->
{{ item }}
{% endfor %}

<!-- 模板继承 -->
<!-- base.html -->
{% block content %}{% endblock %}

<!-- child.html -->
{% extends "base.html" %}
{% block content %}
<p>内容</p>
{% endblock %}

<!-- 包含 -->
{% include "partial.html" %}

<!-- 宏 -->
{% macro input(name, value='', type='text') %}
<input type="{{ type }}" name="{{ name }}" value="{{ value }}">
{% endmacro %}

{{ input('username') }}
{{ input('password', type='password') }}

表单 (Flask-WTF)

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, SelectField, TextAreaField
from wtforms.validators import DataRequired, Email, Length, EqualTo

class LoginForm(FlaskForm):
username = StringField('用户名', validators=[
DataRequired(message='请输入用户名'),
Length(min=3, max=20, message='长度3-20个字符')
])
password = PasswordField('密码', validators=[
DataRequired(message='请输入密码')
])
remember = BooleanField('记住我')
submit = SubmitField('登录')

class RegisterForm(FlaskForm):
email = StringField('邮箱', validators=[Email()])
password = PasswordField('密码', validators=[Length(min=6)])
confirm = PasswordField('确认密码', validators=[
EqualTo('password', message='密码不一致')
])
submit = SubmitField('注册')

# 视图中使用
@app.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
# 验证通过,处理数据
username = form.username.data
return redirect(url_for('index'))
return render_template('login.html', form=form)

# 模板中使用
{{ form.hidden_tag() }}
{{ form.username.label }}
{{ form.username(class_='form-control') }}
{% for error in form.username.errors %}
<span class="error">{{ error }}</span>
{% endfor %}

数据库 (Flask-SQLAlchemy)

配置与模型

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 定义模型
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

# 关系
posts = db.relationship('Post', backref='author', lazy='dynamic')

def __repr__(self):
return f'<User {self.username}>'

class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100))
content = db.Column(db.Text)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

CRUD 操作

# 创建表
db.create_all()
db.drop_all()

# 创建记录
user = User(username='john', email='[email protected]')
db.session.add(user)
db.session.commit()

# 批量创建
db.session.add_all([user1, user2, user3])
db.session.commit()

# 查询
User.query.all() # 所有记录
User.query.get(1) # 按 ID(主键)
User.query.get_or_404(1) # 不存在返回 404
User.query.first() # 第一条
User.query.filter_by(username='john').first() # 条件查询
User.query.filter(User.username == 'john').first()
User.query.filter(User.id > 5).all() # 复杂条件
User.query.filter(User.name.like('%john%')).all()
User.query.filter(User.id.in_([1, 2, 3])).all()
User.query.order_by(User.created_at.desc()).all() # 排序
User.query.limit(10).all() # 限制数量
User.query.paginate(page=1, per_page=10) # 分页

# 更新
user = User.query.get(1)
user.username = 'new_name'
db.session.commit()

# 批量更新
User.query.filter_by(active=False).update({'status': 'deleted'})
db.session.commit()

# 删除
user = User.query.get(1)
db.session.delete(user)
db.session.commit()

# 回滚
db.session.rollback()

用户认证 (Flask-Login)

from flask_login import LoginManager, UserMixin, login_user, logout_user
from flask_login import login_required, current_user

login_manager = LoginManager(app)
login_manager.login_view = 'login'
login_manager.login_message = '请先登录'

# 用户模型
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
# ...

@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))

# 登录
@app.route('/login', methods=['GET', 'POST'])
def login():
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user and check_password_hash(user.password, form.password.data):
login_user(user, remember=form.remember.data)
next_page = request.args.get('next')
return redirect(next_page or url_for('index'))
return render_template('login.html', form=form)

# 登出
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('index'))

# 保护路由
@app.route('/profile')
@login_required
def profile():
return f'Hello {current_user.username}'

# 角色检查
from functools import wraps

def role_required(role):
def decorator(f):
@wraps(f)
@login_required
def decorated(*args, **kwargs):
if current_user.role != role:
abort(403)
return f(*args, **kwargs)
return decorated
return decorator

@app.route('/admin')
@role_required('admin')
def admin():
return 'Admin panel'
from flask import session

# 会话(服务端存储,使用 Cookie 作为标识)
session['user_id'] = 123
session.permanent = True # 持久会话
value = session.get('user_id')
session.pop('user_id', None)
session.clear()

# Cookie(客户端存储)
from flask import make_response

response = make_response('Set cookie')
response.set_cookie('name', 'value', max_age=3600, httponly=True, secure=True)
response.delete_cookie('name')

# 读取 Cookie
name = request.cookies.get('name')

消息闪现

from flask import flash, get_flashed_messages

# 视图中设置
flash('操作成功!', 'success')
flash('出错了!', 'error')
flash('注意!', 'warning')

# 模板中显示
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="alert alert-{{ category }}">{{ message }}</div>
{% endfor %}
{% endif %}
{% endwith %}

# 只获取特定类别
{% with errors = get_flashed_messages(category_filter=['error']) %}
{% for error in errors %}
<div class="alert alert-danger">{{ error }}</div>
{% endfor %}
{% endwith %}

蓝图 (Blueprints)

定义蓝图

# auth/routes.py
from flask import Blueprint, render_template

auth_bp = Blueprint('auth', __name__,
url_prefix='/auth',
template_folder='templates',
static_folder='static')

@auth_bp.route('/login')
def login():
return render_template('auth/login.html')

@auth_bp.route('/register')
def register():
return render_template('auth/register.html')

注册蓝图

# app.py
from flask import Flask
from auth.routes import auth_bp

app = Flask(__name__)
app.register_blueprint(auth_bp)
app.register_blueprint(api_bp, url_prefix='/api/v1')

配置管理

# 基本配置
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'your-secret-key'

# 从对象加载
class Config:
DEBUG = True
SECRET_KEY = 'secret-key'
SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'

app.config.from_object(Config)

# 从环境变量
app.config.from_envvar('APP_CONFIG')

# 从文件
app.config.from_pyfile('config.py')

# 环境区分
import os
env = os.environ.get('FLASK_ENV', 'development')

if env == 'production':
app.config.from_object('config.ProductionConfig')
else:
app.config.from_object('config.DevelopmentConfig')

错误处理

@app.errorhandler(404)
def not_found(error):
if request.path.startswith('/api/'):
return jsonify({'error': 'Not found'}), 404
return render_template('404.html'), 404

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

@app.errorhandler(403)
def forbidden(error):
return render_template('403.html'), 403

# 自定义错误
class ValidationError(Exception):
pass

@app.errorhandler(ValidationError)
def validation_error(error):
return jsonify({'error': str(error)}), 400

请求钩子

# 请求前执行
@app.before_request
def before_request():
# 检查用户状态、记录日志等
g.start_time = time.time()

# 请求后执行(必须返回响应)
@app.after_request
def after_request(response):
# 添加响应头、记录日志等
response.headers['X-Response-Time'] = time.time() - g.start_time
return response

# 请求结束时执行(无论是否有异常)
@app.teardown_request
def teardown_request(exception=None):
# 清理资源
pass

# 应用上下文相关
@app.before_first_request
def before_first_request():
# 仅在第一个请求前执行一次
pass

RESTful API

from flask import Blueprint, jsonify, request

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

@api_bp.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
return jsonify([u.to_dict() for u in users])

@api_bp.route('/users/<int:id>', methods=['GET'])
def get_user(id):
user = User.query.get_or_404(id)
return jsonify(user.to_dict())

@api_bp.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
user = User(**data)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201

@api_bp.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
user = User.query.get_or_404(id)
data = request.get_json()
for key, value in data.items():
setattr(user, key, value)
db.session.commit()
return jsonify(user.to_dict())

@api_bp.route('/users/<int:id>', methods=['DELETE'])
def delete_user(id):
user = User.query.get_or_404(id)
db.session.delete(user)
db.session.commit()
return '', 204

异步支持 (Flask 3.0+)

# Flask 3.0+ 支持异步视图
@app.route('/async')
async def async_view():
data = await fetch_data()
return jsonify(data)

# 异步数据库操作
@app.route('/users')
async def get_users():
users = await db.session.execute(select(User))
return jsonify([u.to_dict() for u in users.scalars()])

# 异步 HTTP 请求
import httpx

async def fetch_data():
async with httpx.AsyncClient() as client:
response = await client.get('https://api.example.com/data')
return response.json()

Blueprint 速查

from flask import Blueprint

# 创建 Blueprint
bp = Blueprint(
'name', # Blueprint 名称
__name__, # 模块名
url_prefix='/prefix', # URL 前缀
subdomain='api', # 子域名
template_folder='templates', # 模板目录
static_folder='static', # 静态文件目录
static_url_path='/static' # 静态文件 URL 路径
)

# 定义路由
@bp.route('/path')
def view():
pass

# 注册 Blueprint
app.register_blueprint(bp)

# 带选项注册
app.register_blueprint(bp, url_prefix='/custom')

# URL 生成
url_for('bp_name.view') # 完整端点名
url_for('.view') # 相对端点(在 Blueprint 内)

# Blueprint 钩子
@bp.before_request
def before():
pass

@bp.after_request
def after(response):
return response

@bp.errorhandler(404)
def not_found(error):
return 'Not Found', 404

# 嵌套 Blueprint
parent = Blueprint('parent', __name__)
child = Blueprint('child', __name__)
parent.register_blueprint(child)
app.register_blueprint(parent)
# 端点:parent.child.view
# URL:/parent/child/path

配置管理速查

# 基本配置
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = 'your-secret-key'

# 从对象加载
class Config:
DEBUG = True
SECRET_KEY = 'secret-key'

app.config.from_object(Config)
app.config.from_object('config.ProductionConfig')

# 从环境变量
app.config.from_envvar('APP_CONFIG')

# 从 Python 文件
app.config.from_pyfile('config.py')

# 从 JSON/TOML 文件
import json
app.config.from_file('config.json', load=json.load)

# 从带前缀的环境变量
app.config.from_prefixed_env('FLASK')
# FLASK_SECRET_KEY -> SECRET_KEY

# 环境区分
import os
env = os.environ.get('FLASK_ENV', 'development')
app.config.from_object(f'config.{env.capitalize()}Config')

# Instance 文件夹
app = Flask(__name__, instance_relative_config=True)
app.config.from_pyfile('config.py', silent=True)

常用配置项

# 安全配置
SECRET_KEY = 'your-secret-key' # 会话签名密钥
SESSION_COOKIE_SECURE = True # 仅 HTTPS
SESSION_COOKIE_HTTPONLY = True # 禁止 JS 访问
SESSION_COOKIE_SAMESITE = 'Lax' # CSRF 防护
PERMANENT_SESSION_LIFETIME = 3600 # 会话有效期(秒)

# 数据库配置
SQLALCHEMY_DATABASE_URI = 'sqlite:///app.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'pool_recycle': 3600,
'pool_pre_ping': True
}

# 文件上传
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 最大 16MB
UPLOAD_FOLDER = 'uploads'

# JSON 配置
JSON_SORT_KEYS = False
JSONIFY_PRETTYPRINT_REGULAR = False

# 服务器配置
SERVER_NAME = 'example.com' # 用于 URL 生成
APPLICATION_ROOT = '/' # 应用挂载路径
PREFERRED_URL_SCHEME = 'https' # 首选 URL 协议

部署命令速查

# Gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 run:app
gunicorn -k gevent -w 4 run:app # gevent 模式
gunicorn -c gunicorn.conf.py run:app # 使用配置文件

# uWSGI
uwsgi --http :8000 --wsgi-file run.py --callable app
uwsgi --ini uwsgi.ini

# Waitress (Windows 支持)
waitress-serve --port=8000 run:app

# Flask 开发服务器(仅开发)
flask run
flask run --host=0.0.0.0 --port=8000
flask run --debug

# 数据库迁移
flask db init # 初始化
flask db migrate -m "message" # 生成迁移
flask db upgrade # 应用迁移
flask db downgrade # 回滚

# Docker
docker build -t myapp .
docker run -p 8000:8000 myapp
docker-compose up -d

常用扩展

扩展用途安装命令
Flask-SQLAlchemy数据库 ORMpip install flask-sqlalchemy
Flask-WTF表单处理pip install flask-wtf
Flask-Login用户认证pip install flask-login
Flask-Migrate数据库迁移pip install flask-migrate
Flask-Mail邮件发送pip install flask-mail
Flask-Caching缓存支持pip install flask-caching
Flask-RESTfulREST APIpip install flask-restful
Flask-CORS跨域支持pip install flask-cors
Flask-JWT-ExtendedJWT 认证pip install flask-jwt-extended
Flask-Limiter请求限制pip install flask-limiter
Flask-Admin后台管理pip install flask-admin
Flask-RESTXAPI 文档pip install flask-restx
Flask-Marshmallow序列化pip install flask-marshmallow

常用命令

# 运行开发服务器
flask run
flask run --host=0.0.0.0 --port=8080
flask run --debug

# 设置环境变量
export FLASK_APP=app.py
export FLASK_ENV=development
export FLASK_DEBUG=1

# 进入 Flask Shell
flask shell

# 数据库迁移 (Flask-Migrate)
flask db init # 初始化
flask db migrate -m "message" # 生成迁移
flask db upgrade # 应用迁移
flask db downgrade # 回滚迁移
flask db current # 当前版本
flask db history # 迁移历史

# 自定义命令
flask --help

最佳实践速查

# 1. 安全的密钥
import secrets
app.config['SECRET_KEY'] = secrets.token_hex(32)

# 2. 环境变量配置
from dotenv import load_dotenv
load_dotenv()

# 3. 分页
from flask import request
page = request.args.get('page', 1, type=int)
pagination = User.query.paginate(page=page, per_page=20)

# 4. JSON 响应统一格式
def api_response(data=None, message='success', code=200):
return jsonify({
'code': code,
'message': message,
'data': data
}), code

# 5. CORS 配置
from flask_cors import CORS
CORS(app, resources={r"/api/*": {"origins": "*"}})

# 6. 请求日志
import logging
logging.basicConfig(level=logging.INFO)

# 7. 性能优化 - 缓存
from flask_caching import Cache
cache = Cache(app, config={'CACHE_TYPE': 'redis'})

@app.route('/expensive')
@cache.cached(timeout=300)
def expensive_operation():
return compute_result()