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'
会话与 Cookie
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 | 数据库 ORM | pip 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-RESTful | REST API | pip install flask-restful |
| Flask-CORS | 跨域支持 | pip install flask-cors |
| Flask-JWT-Extended | JWT 认证 | pip install flask-jwt-extended |
| Flask-Limiter | 请求限制 | pip install flask-limiter |
| Flask-Admin | 后台管理 | pip install flask-admin |
| Flask-RESTX | API 文档 | 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()