Skip to content

Flask 速查表

轻量级 Python Web 框架(微框架)。

01

入门

第一个应用

Flask 是一个微框架——它提供基础功能,让你按需添加。debug=True 启用自动重载和详细错误页面。

flask
# install Flask
pip install flask

# app.py
from flask import Flask, render_template

app = Flask(__name__)

@app.route("/")
def home():
    return render_template("index.html")

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

运行开发服务器

切勿在生产环境使用内置服务器。FLASK_APP 指向包含应用实例的模块。使用 --host=0.0.0.0 使其在网络中可访问。

flask
# Set the entry point and enable debug mode
export FLASK_APP=app.py
export FLASK_ENV=development
flask run

# Override host and port
flask run --host=0.0.0.0 --port=5000

# Or run the module directly
python app.py

虚拟环境

虚拟环境将项目依赖与系统 Python 隔离。requirements.txt 记录确切版本,他人可通过 pip install -r requirements.txt 复现环境。

flask
# Create an isolated environment
python -m venv venv

# Activate (Linux/macOS)
source venv/bin/activate

# Activate (Windows PowerShell)
venv\Scripts\Activate.ps1

# Install dependencies
pip install flask
pip freeze > requirements.txt

项目结构

Flask 不强制布局,但将静态文件、模板和配置分开是惯例。instance/ 文件夹存放不应纳入版本控制的部署特定密钥。

flask
myapp/
  app.py            # application factory or instance
  config.py         # configuration classes
  requirements.txt  # pinned dependencies
  instance/
    config.py       # secret instance config (not in VCS)
  static/           # CSS, JS, images
    css/style.css
  templates/        # Jinja2 templates
    base.html
    index.html
  models.py         # database models
  views.py          # route handlers

应用工厂模式

工厂模式将应用创建推迟到函数调用时,这对多实例、测试和扩展至关重要。Flask 的 --app 选项接受返回应用的回调函数。

flask
from flask import Flask

def create_app(config_name="default"):
    app = Flask(__name__)
    app.config.from_object(f"config.{config_name.title()}Config")

    from .views import bp as views_bp
    app.register_blueprint(views_bp)

    @app.route("/health")
    def health():
        return {"status": "ok"}

    return app

# Run with: flask --app myapp create_app run
# or export FLASK_APP=myapp:create_app

最小应用结构

传入 __name__ 让 Flask 知道去哪里找静态文件和模板。如果布局与默认不同,可以用构造函数参数覆盖文件夹位置。

flask
from flask import Flask

app = Flask(
    __name__,
    static_folder="static",
    template_folder="templates",
    static_url_path="/static",
)

# __name__ lets Flask locate resources relative to the module.
# Route decorators map URL patterns to view functions.
@app.route("/")
def index():
    return "Hello, Flask!"
02

路由

变量规则

用 <converter:varname> 标记的部分捕获 URL 片段。内置转换器:string(默认)、int、float、path(接受斜杠)、uuid。值作为函数参数传入。

flask
from flask import Flask

app = Flask(__name__)

@app.route("/user/<username>")
def show_user(username):
    return f"User: {username}"

@app.route("/post/<int:post_id>")
def show_post(post_id):
    return f"Post #{post_id}"

@app.route("/path/<path:subpath>")
def show_path(subpath):
    return f"Path: {subpath}"

URL 转换器

转换器验证并转换 URL 变量。在 app.url_map.converters 上注册自定义 BaseConverter 子类可强制正则模式,如产品代码。

flask
@app.route("/item/<int:item_id>")
def item_int(item_id):
    return f"Integer ID: {item_id}"

@app.route("/weight/<float:kg>")
def weight(kg):
    return f"{kg} kg"

@app.route("/uid/<uuid:token>")
def by_token(token):
    return str(token)

# Custom converter
from werkzeug.routing import BaseConverter

class RegexConverter(BaseConverter):
    def __init__(self, url_map, *items):
        super().__init__(url_map)
        self.regex = items[0]

app.url_map.converters["re"] = RegexConverter

@app.route("/code/<re:[a-z]{3}-[0-9]{4}:code>")
def code(code):
    return code

唯一 URL 与重定向

尾部斜杠很重要。以 '/' 结尾的路由会重定向无斜杠形式;不以 '/' 结尾的路由对斜杠形式返回 404。每个路由选择一种约定并保持一致。

flask
@app.route("/projects/")
def projects():
    return "Projects page"

# /projects/  -> 200 OK
# /projects   -> 301 redirect to /projects/

@app.route("/about")
def about():
    return "About page"

# /about    -> 200 OK
# /about/   -> 404 Not Found

HTTP 方法

使用 methods 参数接受多种动词,或用 @app.get / @app.post 快捷方式(Flask 2.0+)处理单方法路由。GET 是默认方法。

flask
from flask import request, render_template

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        return do_login()
    return render_template("login.html")

# Method-specific shortcuts
@app.get("/profile")
def profile_get():
    return render_template("profile.html")

@app.post("/profile")
def profile_post():
    return update_profile()

用 url_for 构建 URL

url_for 从端点名称(默认为视图函数名)生成 URL。它能适应 URL 变更,传入 _external=True 时生成绝对 URL。

flask
from flask import Flask, url_for

app = Flask(__name__)

@app.route("/")
def index():
    # builds '/user/alice' instead of hardcoding
    return f'<a href="{url_for("show_user", username="alice")}">Alice</a>'

@app.route("/user/<username>")
def show_user(username):
    return f"Hello {username}"

with app.test_request_context():
    print(url_for("show_user", username="bob"))  # /user/bob

一个视图的多条规则

add_url_rule 不用装饰器注册 URL。通过给每条规则不同的端点但相同的视图函数和默认值,一个处理器可服务多种 URL 形式。

flask
from flask import Flask

app = Flask(__name__)

def show(id=None, name=None):
    if id is not None:
        return f"By id: {id}"
    return f"By name: {name}"

app.add_url_rule("/by-id/<int:id>", "by_id", show, defaults={"name": None})
app.add_url_rule("/by-name/<name>", "by_name", show, defaults={"id": None})

# Two endpoints share one function with different defaults.
03

请求与响应

请求对象

request 是绑定到当前请求上下文的线程局部代理。它暴露 URL、方法、头部、主体和客户端信息,无需显式传参。

flask
from flask import Flask, request

app = Flask(__name__)

@app.route("/info")
def info():
    return {
        "url": request.url,
        "method": request.method,
        "remote_addr": request.remote_addr,
        "user_agent": str(request.user_agent),
        "headers": dict(request.headers),
    }

创建响应

make_response 完全控制状态、头部和主体。简单情况下,返回字符串、字典(JSON)或 (body, status, headers) 元组。

flask
from flask import Flask, make_response

app = Flask(__name__)

@app.route("/")
def index():
    resp = make_response("Hello, World!")
    resp.status_code = 201
    resp.headers["X-Custom"] = "yes"
    resp.set_cookie("visited", "1")
    return resp

# Returning a tuple: (body, status, headers)
@app.route("/quick")
def quick():
    return "OK", 202, {"X-Quick": "1"}

响应对象

需要自定义 mimetype(如 XML、CSV)或头部时使用 Response。将 Content-Disposition 设为 attachment 可强制浏览器下载。

flask
from flask import Response

@app.route("/xml")
def xml():
    return Response("<msg>hi</msg>", mimetype="application/xml")

@app.route("/text")
def text():
    return Response("plain text", mimetype="text/plain")

@app.route("/csv")
def csv():
    body = "a,b,c\n1,2,3\n"
    headers = {"Content-Disposition": "attachment; filename=data.csv"}
    return Response(body, mimetype="text/csv", headers=headers)

自定义状态码与头部

返回 (body, status, headers) 元组可简洁地自定义响应。用 204 No Content 表示成功但无主体,使用标准 HTTP 状态码表达语义。

flask
@app.route("/created")
def created():
    return "done", 201, {"Location": "/resource/1"}

@app.route("/no-content")
def no_content():
    return "", 204

@app.route("/teapot")
def teapot():
    return "I'm a teapot", 418

流式响应

生成器返回值将数据流式传输给客户端。适用于大文件(避免整个文件加载到内存)和 Server-Sent Events。需设置适当的 mimetype。

flask
from flask import Response

def generate():
    yield "data: first\n\n"
    yield "data: second\n\n"
    yield "data: third\n\n"

@app.route("/stream")
def stream():
    return Response(generate(), mimetype="text/event-stream")

@app.route("/big-file")
def big_file():
    def chunks():
        with open("large.log", "rb") as f:
            while chunk := f.read(8192):
                yield chunk
    return Response(chunks(), mimetype="application/octet-stream")

响应中的 Cookie

set_cookie 写入 Set-Cookie 头;delete_cookie 使其过期。始终设置 httponly 和 samesite 以保证安全。Cookie 是响应的一部分,不是请求主体。

flask
from flask import make_response

@app.route("/set-theme")
def set_theme():
    resp = make_response("theme set")
    resp.set_cookie("theme", "dark", max_age=30*86400, httponly=True, samesite="Lax")
    return resp

@app.route("/clear-theme")
def clear_theme():
    resp = make_response("cleared")
    resp.delete_cookie("theme")
    return resp
04

请求对象(form/args/files)

查询参数(args)

request.args 是查询字符串值的 MultiDict。使用 get 带默认值和类型转换,用 getlist 处理重复键。切勿直接索引——缺失键会引发 400。

flask
from flask import request

@app.route("/search")
def search():
    q = request.args.get("q", "")           # default ""
    page = request.args.get("page", 1, type=int)
    tags = request.args.getlist("tag")       # /search?tag=a&tag=b -> ["a","b"]
    return {"q": q, "page": page, "tags": tags}

表单数据

request.form 保存 POST 的 application/x-www-form-urlencoded 或 multipart 数据。用 .get() 避免可选字段的 KeyError。GET 请求的 form 为空。

flask
from flask import request, render_template

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")
        if not username or not password:
            return "Missing fields", 400
        return f"Welcome {username}"
    return render_template("login.html")

文件上传

request.files 是 FileStorage 对象的 MultiDict。保存前务必通过 secure_filename 处理用户提供的文件名,以去除路径分隔符和危险字符。

flask
from flask import request
from werkzeug.utils import secure_filename
import os

@app.route("/upload", methods=["POST"])
def upload():
    file = request.files.get("file")
    if not file or file.filename == "":
        return "No file", 400
    name = secure_filename(file.filename)
    file.save(os.path.join("uploads", name))
    return "saved"

JSON 主体

get_json() 仅在 Content-Type 为 application/json 时解析主体为 JSON。force=True 忽略头部(谨慎使用)。silent=True 抑制解析错误。

flask
from flask import request

@app.route("/api/echo", methods=["POST"])
def echo():
    data = request.get_json()        # returns None if not JSON
    if data is None:
        return "Expected JSON", 415
    return {"you_sent": data}

# Force parsing even without the right header (risky)
@app.route("/api/force")
def force():
    data = request.get_json(force=True)
    return data

头部与 Cookie

request.headers 是类字典的 EnvironHeaders 对象;request.cookies 保存客户端发送的 Cookie。两者均为只读——在响应上设置头部/Cookie。

flask
from flask import request

@app.route("/inspect")
def inspect():
    auth = request.headers.get("Authorization")
    accept = request.headers.get("Accept")
    session_id = request.cookies.get("session_id")
    return {
        "auth": auth,
        "accept": accept,
        "session_id": session_id,
    }

请求方法与属性

request 暴露连接和主体的元数据。is_json 检查 Content-Type,is_secure 检查 HTTPS,query_string 是 '?' 后的原始字节。

flask
from flask import request

@app.route("/state")
def state():
    return {
        "method": request.method,
        "is_xhr": request.is_json,
        "content_type": request.content_type,
        "content_length": request.content_length,
        "is_secure": request.is_secure,
        "scheme": request.scheme,
        "host": request.host,
        "path": request.path,
        "query_string": request.query_string.decode(),
    }
05

Jinja2 模板

渲染模板

render_template 从 templates/ 文件夹加载 Jinja2 文件,并将关键字参数注入为变量。{{ }} 语法输出表达式;Flask 自动转义 HTML 以保证安全。

flask
from flask import render_template

@app.route("/<name>")
def hello(name):
    return render_template("hello.html", name=name, title="Welcome")

# templates/hello.html
# <!DOCTYPE html>
# <html>
#   <head><title>{{ title }}</title></head>
#   <body>
#     <h1>Hello {{ name }}!</h1>
#   </body>
# </html>

模板变量

Jinja2 支持属性(user.name)和索引(user['age'])访问。safe 过滤器将字符串标记为可信以避免转义——仅用于你完全控制的内容。

flask
# View
@app.route("/")
def index():
    return render_template(
        "index.html",
        user={"name": "Alice", "age": 30},
        items=["apple", "banana", "cherry"],
        html_content="<b>safe?</b>",
    )

# Template
# <p>Name: {{ user.name }}</p>
# <p>Age: {{ user["age"] }}</p>
# <p>First: {{ items[0] }}</p>
# <p>Raw HTML: {{ html_content | safe }}</p>

控制结构

Jinja2 支持 {% if %}、{% for %} 和 {% block %}。循环内特殊的 loop 变量提供 index、index0、first、last、length 和 revindex。for-else 在可迭代对象为空时执行。

flask
{# if / elif / else #}
{% if user.age >= 18 %}
  Adult
{% elif user.age >= 13 %}
  Teen
{% else %}
  Child
{% endif %}

{# for loop #}
<ul>
{% for item in items %}
  <li>{{ loop.index }}: {{ item }}</li>
{% else %}
  <li>No items</li>
{% endfor %}
</ul>

模板继承

extends 引入基础模板,子模板覆盖命名块。这让共享布局(导航、脚本)集中在一处。super() 渲染父块的内容。

flask
{# templates/base.html #}
<!DOCTYPE html>
<html>
<head>
  <title>{% block title %}Default{% endblock %}</title>
</head>
<body>
  {% block content %}{% endblock %}
</body>
</html>

{# templates/child.html #}
{% extends "base.html" %}

{% block title %}Home{% endblock %}

{% block content %}
  <h1>Welcome home</h1>
{% endblock %}

过滤器

过滤器通过管道(|)语法转换值并接受参数。在 app.jinja_env.filters 上注册自定义过滤器可在模板间复用逻辑。

flask
{{ name | capitalize }}      {# alice -> Alice #}
{{ price | round(2) }}       {# 3.14159 -> 3.14 #}
{{ items | length }}         {# count #}
{{ text | truncate(20) }}    {# shorten #}
{{ tags | join(", ") }}      {# ["a","b"] -> "a, b" #}
{{ value | default("n/a") }} {# fallback #}
{{ html | striptags }}       {# remove HTML tags #}

{# Custom filter #}
def reverse_filter(s):
    return s[::-1]

app.jinja_env.filters["reverse"] = reverse_filter
{{ "hello" | reverse }}  {# olleh #}

宏是可复用的模板片段,类似函数。用 {% from %} 或 {% import %} 导入。它们让重复的 HTML(表单字段、卡片)保持 DRY。

flask
{# templates/macros.html #}
{% macro input(name, value="", type="text") %}
  <input type="{{ type }}" name="{{ name }}" value="{{ value }}">
{% endmacro %}

{% macro label(text, for_id) %}
  <label for="{{ for_id }}">{{ text }}</label>
{% endmacro %}

{# In another template #}
{% from "macros.html" import input, label %}

<form>
  {{ label("Username", "u") }}
  {{ input("username") }}
  {{ label("Password", "p") }}
  {{ input("password", type="password") }}
</form>
06

静态文件

提供静态文件

Flask 自动为 static 文件夹注册 /static/<path:filename> 路由。始终用 url_for 构建 URL,即使 static_url_path 改变也能保持路径正确。

flask
# Files in the static/ folder are served at /static/<filename>
# static/css/style.css -> http://localhost:5000/static/css/style.css

# Reference them in templates
# <link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
# <script src="{{ url_for('static', filename='js/app.js') }}"></script>
# <img src="{{ url_for('static', filename='img/logo.png') }}">

静态资源的 url_for

url_for('static', filename=...) 解析静态文件路径。传入 _external=True 获取绝对 URL(用于邮件或订阅源),传入查询参数用于缓存清除。

flask
from flask import Flask, url_for

app = Flask(__name__)

with app.test_request_context():
    print(url_for("static", filename="css/style.css"))
    # /static/css/style.css
    print(url_for("static", filename="js/app.js", _external=True))
    # http://localhost/static/js/app.js

自定义静态文件夹

覆盖 static_folder(磁盘位置)和 static_url_path(URL 前缀)以匹配项目约定。默认是 folder='static' 映射到 /static。

flask
from flask import Flask

app = Flask(
    __name__,
    static_folder="assets",       # folder on disk
    static_url_path="/public",    # URL prefix
)

# Now files in assets/ are served at /public/<filename>
# Template: url_for('static', filename='css/style.css') -> /public/css/style.css

Favicon

浏览器默认请求根路径的 /favicon.ico。添加显式路由或通过 <link> 标签链接图标。send_from_directory 安全地从目录提供文件。

flask
from flask import send_from_directory
import os

@app.route("/favicon.ico")
def favicon():
    return send_from_directory(
        os.path.join(app.root_path, "static"),
        "favicon.ico",
        mimetype="image/vnd.microsoft.icon",
    )

# Or place favicon.ico in static/ and link it:
# <link rel="icon" href="{{ url_for('static', filename='favicon.ico') }}">

缓存清除

浏览器按 URL 缓存静态资源。追加版本哈希可强制客户端在文件更改后获取新副本。构建工具(Vite、Flask-Assets)可在大型项目中自动化此过程。

flask
# Append a version query param to force reloads
# <link href="{{ url_for('static', filename='css/app.css', v='1.2') }}">

# Or compute a hash at startup
import hashlib, os

def asset_version(filename):
    path = os.path.join(app.static_folder, filename)
    with open(path, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()[:8]

@app.template_filter("bust")
def bust(filename):
    v = asset_version(filename)
    return url_for("static", filename=filename) + f"?v={v}"

# Template: <link href="{{ 'css/app.css' | bust }}">
07

蓝图

创建蓝图

蓝图将相关路由、模板和静态文件分组到一个模块。第一个参数是蓝图名称;url_prefix 为其定义的每个路由前置一个路径。

flask
# auth.py
from flask import Blueprint, render_template

bp = Blueprint("auth", __name__, url_prefix="/auth")

@bp.route("/login")
def login():
    return render_template("auth/login.html")

@bp.route("/register")
def register():
    return render_template("auth/register.html")

注册蓝图

register_blueprint 将蓝图附加到应用。可在注册时覆盖 url_prefix,使同一蓝图在不同应用中挂载到不同路径。

flask
from flask import Flask
from auth import bp as auth_bp
from blog import bp as blog_bp

app = Flask(__name__)
app.register_blueprint(auth_bp)           # /auth/*
app.register_blueprint(blog_bp, url_prefix="/blog")  # /blog/*

# Register later, or with a name override
app.register_blueprint(auth_bp, name="alt_auth")

带 URL 前缀的蓝图

在蓝图上或注册时定义 url_prefix。对 API 路径进行版本控制(/api/v1)可在将来引入 /api/v2 而不破坏现有客户端。

flask
# api/__init__.py
from flask import Blueprint

bp = Blueprint("api", __name__, url_prefix="/api/v1")

@bp.route("/users")
def list_users():
    return {"users": []}

@bp.route("/users/<int:uid>")
def get_user(uid):
    return {"id": uid}

# All routes become /api/v1/users...

蓝图资源

蓝图可携带自己的模板和静态文件夹。模板解析先检查蓝图,再检查应用,因此蓝图本地文件可覆盖同名的应用级文件。

flask
bp = Blueprint(
    "admin",
    __name__,
    url_prefix="/admin",
    template_folder="templates",   # blueprint-local templates
    static_folder="static",        # blueprint-local static
)

# Templates: blueprint lookups first, then app templates
# render_template("admin/dashboard.html")
# Static: url_for('admin.static', filename='css/admin.css')

蓝图错误处理器

蓝图级错误处理器仅捕获该蓝图路由引发的错误,提供特定分区的错误页面。应用级处理器处理未捕获的内容。

flask
from flask import Blueprint, render_template

bp = Blueprint("blog", __name__)

@bp.errorhandler(404)
def blog_not_found(e):
    return render_template("blog/404.html"), 404

@bp.errorhandler(403)
def blog_forbidden(e):
    return render_template("blog/403.html"), 403

嵌套蓝图

Flask 2.0+ 支持嵌套蓝图。在父蓝图上用 register_blueprint 注册子蓝图;它们的 url_prefix 会拼接,让你用小模块组合大型应用。

flask
from flask import Blueprint

parent = Blueprint("parent", __name__, url_prefix="/parent")
child = Blueprint("child", __name__, url_prefix="/child")

@child.route("/info")
def info():
    return "nested info"

parent.register_blueprint(child)
app.register_blueprint(parent)

# Final URL: /parent/child/info
08

上下文(g/session)

应用上下文

应用上下文使 current_app 和 g 在请求外可用(如 CLI 命令或后台任务中)。在请求外工作时用 app.app_context() 推入上下文。

flask
from flask import current_app, g

@app.route("/")
def index():
    app_name = current_app.name
    config_value = current_app.config["SECRET_KEY"]
    return f"Running {app_name}"

# Push a context manually (scripts, shell)
with app.app_context():
    print(current_app.config["DEBUG"])

g 对象

g 是每个请求的命名空间,用于存储数据库连接等共享资源。它在每次请求时重置,并在 teardown_appcontext 处理器中清理。

flask
from flask import g
import sqlite3

def get_db():
    if "db" not in g:
        g.db = sqlite3.connect("app.db")
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(exc):
    db = g.pop("db", None)
    if db is not None:
        db.close()

请求上下文

请求上下文将 request、session 和 url_for 绑定到线程。在测试和脚本中使用 test_request_context 模拟请求,无需 HTTP 服务器。

flask
from flask import request

# Flask pushes a request context automatically during dispatch.
# Push one manually to use request/url_for outside a view:
with app.test_request_context("/?q=flask"):
    print(request.args.get("q"))   # flask
    print(request.path)            # /

# request_context for real WSGI environments
from werkzeug.test import EnvironBuilder
env = EnvironBuilder("/post/1").get_environ()
with app.request_context(env):
    print(request.path)

current_app 与 request 代理

current_app、request、session 和 g 是 LocalProxy 对象,将属性访问转发到活动上下文。切勿将它们捕获到全局变量中——始终在调用时访问。

flask
from flask import current_app, request, session, g

# These are LocalProxy objects, not the real thing.
# They resolve to the current context's object at access time.

@app.route("/")
def index():
    current_app.logger.info("hit /")
    g.request_start = time.time()
    session["visited"] = True
    return current_app.config["APP_NAME"]

# Avoid storing proxies in module-level variables;
# always access them inside a context.

Session 对象

session 是基于签名 Cookie 的字典。设置 app.secret_key 让 Flask 对其签名。值存储在客户端且用户可读取(但不可篡改),因此切勿存储密钥。

flask
from flask import session

@app.route("/login")
def login():
    session["user_id"] = 42
    session["role"] = "admin"
    return "logged in"

@app.route("/who")
def who():
    uid = session.get("user_id")
    return f"user {uid}" if uid else "anonymous"

@app.route("/logout")
def logout():
    session.clear()
    return "logged out"

上下文生命周期

理解生命周期是正确使用钩子的关键。teardown_* 处理器始终运行(即使在出错时),因此是关闭数据库连接和释放资源的正确位置。

flask
# Per-request lifecycle:
# 1. Application context pushed
# 2. Request context pushed
# 3. before_request handlers run
# 4. View function runs
# 5. after_request handlers run (can modify response)
# 6. Request context popped
# 7. teardown_request handlers run
# 8. teardown_appcontext handlers run (close resources)

@app.before_request
def before():
    g.start = time.time()

@app.teardown_request
def teardown(exc):
    elapsed = time.time() - getattr(g, "start", time.time())
    app.logger.debug("request took %.3fs", elapsed)
09

SQLAlchemy 集成

安装 Flask-SQLAlchemy

Flask-SQLAlchemy 封装 SQLAlchemy 并与 Flask 的应用上下文集成。设置 SQLALCHEMY_DATABASE_URI 为连接字符串,禁用 track modifications 以节省内存。

flask
pip install flask-sqlalchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

db = SQLAlchemy(app)

# In a factory, init with db.init_app(app)

定义模型

模型继承 db.Model;列是 db.Column 实例。常见类型包括 Integer、String、Text、DateTime、Boolean。用 unique=True 和 nullable=False 添加约束。

flask
from datetime import datetime

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)

    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(200), nullable=False)
    body = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"))

查询数据

Model.query 是便捷查询接口。filter_by(keyword=value) 简单;filter(expression) 支持 >、<、like、in_ 等操作符。paginate() 返回条目和分页元数据。

flask
# Fetch all
users = User.query.all()

# Primary key lookup
user = User.query.get(1)

# Filter
admins = User.query.filter_by(role="admin").all()
active = User.query.filter(User.email.isnot(None)).first()

# Order, limit, paginate
recent = User.query.order_by(User.created_at.desc()).limit(10).all()
page = User.query.paginate(page=2, per_page=20)

# Count
total = User.query.count()

添加与更新

所有更改通过 db.session 在事务中进行。add 暂存新对象;commit 写入。出错时用 db.session.rollback() 回滚以保持会话可用。

flask
# Create
u = User(username="alice", email="[email protected]")
db.session.add(u)
db.session.commit()

# Update
u.email = "[email protected]"
db.session.commit()

# Delete
db.session.delete(u)
db.session.commit()

# Bulk insert
db.session.add_all([User(username="b"), User(username="c")])
db.session.commit()

关系

db.relationship 定义模型间的链接。backref 添加反向属性。lazy='dynamic' 返回查询对象而非列表,适合需要进一步过滤的大型集合。

flask
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80))
    posts = db.relationship("Post", backref="author", lazy="dynamic")

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200))
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"))

# Usage
user = User.query.get(1)
user.posts.all()              # list of posts
post = Post.query.get(5)
post.author.username          # backref to user

用 Flask-Migrate 迁移

Flask-Migrate 封装 Alembic 对数据库模式进行版本控制。运行 migrate 从模型变更自动生成脚本,然后 upgrade 应用。始终在提交前审查自动生成的脚本。

flask
pip install flask-migrate

from flask_migrate import Migrate

migrate = Migrate(app, db)

# CLI commands
# flask db init          # create migrations folder
# flask db migrate -m "create users"
# flask db upgrade       # apply migrations
# flask db downgrade     # revert one step
# flask db current       # show current revision
10

WTForms(Flask-WTF)

安装 Flask-WTF

Flask-WTF 将 WTForms 与 Flask 集成,并自动添加 CSRF 保护。需要 SECRET_KEY 以便签名 CSRF 令牌。仅在公共 API 端点禁用 CSRF。

flask
pip install flask-wtf

from flask import Flask
from flask_wtf import FlaskForm

app = Flask(__name__)
app.config["SECRET_KEY"] = "change-me-in-production"
app.config["WTF_CSRF_ENABLED"] = True

# CSRF protection is enabled by default
# when SECRET_KEY is set.

定义表单

表单继承 FlaskForm。每个字段接受标签和在 validate() 时运行的验证器列表。验证器强制必填、长度、邮箱格式等。

flask
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    email = StringField("Email", validators=[DataRequired(), Email()])
    password = PasswordField("Password", validators=[DataRequired(), Length(min=8)])
    submit = SubmitField("Log In")

class RegisterForm(FlaskForm):
    username = StringField("Username", validators=[DataRequired(), Length(min=3, max=20)])
    email = StringField("Email", validators=[DataRequired(), Email()])
    submit = SubmitField("Sign Up")

在模板中渲染表单

form.hidden_tag() 渲染 CSRF 令牌和隐藏字段。每个字段可分别渲染标签、输入和错误。快速输出用 {{ form.field() }} 渲染 HTML 输入。

flask
{# templates/login.html #}
<form method="POST">
  {{ form.hidden_tag() }}  {# CSRF token #}
  <p>
    {{ form.email.label }}
    {{ form.email() }}
    {% for err in form.email.errors %}<span class="err">{{ err }}</span>{% endfor %}
  </p>
  <p>
    {{ form.password.label }}
    {{ form.password() }}
  </p>
  {{ form.submit() }}
</form>

验证表单

validate_on_submit() 仅在 POST 请求且所有验证器通过时返回 True。成功时从 form.field.data 读取值。失败时重新渲染模板显示错误。

flask
from flask import render_template, redirect, url_for, flash

@app.route("/login", methods=["GET", "POST"])
def login():
    form = LoginForm()
    if form.validate_on_submit():   # POST + valid
        email = form.email.data
        flash(f"Logged in as {email}")
        return redirect(url_for("dashboard"))
    return render_template("login.html", form=form)

# validate_on_submit() returns False on GET or invalid POST

内置验证器

验证器是可调用对象,失败时引发 ValidationError。EqualTo 常用于密码确认。在列表中链式调用,按顺序运行直到一个失败。

flask
from wtforms.validators import (
    DataRequired,   # field not empty
    Email,          # valid email format
    Length,         # Length(min=3, max=20)
    NumberRange,    # NumberRange(min=0, max=100)
    URL,            # valid URL
    Regexp,         # Regexp(r"^\\w+$")
    EqualTo,        # EqualTo("password") - match another field
    Optional,       # skip validation if empty
    InputRequired,  # field submitted (even if empty string)
    NoneOf,         # NoneOf(["admin", "root"])
)

password = PasswordField("Password", validators=[DataRequired(), Length(min=8)])
confirm = PasswordField("Confirm", validators=[EqualTo("password")])

FileField

使用 flask_wtf.file 的 FileField 配合 FileRequired 和 FileAllowed 验证上传。上传文件是 form.field.data 处的 FileStorage 对象。表单标签须包含 enctype=multipart/form-data。

flask
from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileRequired, FileAllowed

class UploadForm(FlaskForm):
    image = FileField("Image", validators=[
        FileRequired(),
        FileAllowed(["jpg", "png", "gif"], "Images only!"),
    ])
    submit = SubmitField("Upload")

@app.route("/upload", methods=["GET", "POST"])
def upload():
    form = UploadForm()
    if form.validate_on_submit():
        f = form.image.data
        f.save("uploads/" + f.filename)
        return "uploaded"
    return render_template("upload.html", form=form)
11

Flask-Login 认证

安装 Flask-Login

Flask-Login 管理用户会话:登录、登出和记住用户。设置 login_view 以便未认证用户重定向到那里。login_message 在重定向时闪现。

flask
pip install flask-login

from flask import Flask
from flask_login import LoginManager

app = Flask(__name__)
app.config["SECRET_KEY"] = "secret"

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = "login"
login_manager.login_message = "Please log in to access this page."

用户模型

用户模型必须混入 UserMixin(提供 is_authenticated、is_active、get_id 等)并实现 user_loader 回调按 ID 返回用户。切勿存储明文密码——必须哈希。

flask
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    password_hash = db.Column(db.String(256))

    def set_password(self, pw):
        self.password_hash = generate_password_hash(pw)

    def check_password(self, pw):
        return check_password_hash(self.password_hash, pw)

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

登录视图

login_user 将用户加载到会话。传入 remember=True 获取持久 Cookie。登录前始终通过哈希检查验证密码,失败时闪现通用错误。

flask
from flask import render_template, redirect, url_for, flash
from flask_login import login_user
from forms import LoginForm

@app.route("/login", methods=["GET", "POST"])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user and user.check_password(form.password.data):
            login_user(user, remember=form.remember.data)
            return redirect(url_for("dashboard"))
        flash("Invalid username or password")
    return render_template("login.html", form=form)

保护路由

@login_required 装饰器拒绝匿名用户并重定向到 login_view。current_user 是已登录用户(或 AnonymousUserMixin)的代理,在视图和模板中可用。

flask
from flask_login import login_required, current_user

@app.route("/dashboard")
@login_required
def dashboard():
    return f"Welcome {current_user.username}"

@app.route("/settings")
@login_required
def settings():
    return render_template("settings.html", user=current_user)

# Unauthenticated users redirect to login_view
# @login_required must be the innermost decorator

登出

logout_user 从会话中清除用户。用 @login_required 保护路由,确保只有已登录用户才能登出。之后重定向到公共页面(index 或 login)。

flask
from flask_login import logout_user, login_required
from flask import redirect, url_for, flash

@app.route("/logout")
@login_required
def logout():
    logout_user()
    flash("You have been logged out.")
    return redirect(url_for("index"))

# logout_user() clears the session and cookies

记住我与 current_user

remember=True 设置单独的 Cookie,使用户在关闭浏览器后保持登录。current_user 在视图和模板中均可用,便于根据认证状态条件渲染 UI。

flask
from flask_login import current_user

# In the login view
login_user(user, remember=True)

# remember=True stores a separate long-lived cookie
# so the session survives a browser restart.

# current_user is available everywhere a context exists
@app.route("/")
def index():
    if current_user.is_authenticated:
        return f"Hi {current_user.username}"
    return "Welcome, guest"

# In templates:
# {% if current_user.is_authenticated %}
#   <a href="{{ url_for('logout') }}">Logout</a>
# {% endif %}
12

错误处理

abort()

abort() 立即停止视图并返回 HTTP 错误。传入状态码和可选消息。它引发 HTTPException,Flask 将其转换为匹配的错误响应。

flask
from flask import abort

@app.route("/user/<int:id>")
def get_user(id):
    user = find_user(id)
    if user is None:
        abort(404)            # Not Found
    if not can_access(user):
        abort(403, "Forbidden")   # Forbidden with description
    return render_template("user.html", user=user)

# Common codes: 400 Bad Request, 401 Unauthorized,
# 403 Forbidden, 404 Not Found, 418 Teapot, 500 Server Error

自定义错误页面

用 @app.errorhandler(code) 注册处理器。返回 (template, status_code) 元组。这些处理器捕获 abort() 调用和映射到该状态的异常。

flask
from flask import render_template

@app.errorhandler(404)
def not_found(e):
    return render_template("errors/404.html"), 404

@app.errorhandler(403)
def forbidden(e):
    return render_template("errors/403.html"), 403

@app.errorhandler(500)
def server_error(e):
    return render_template("errors/500.html"), 500

异常错误处理器

可以为自定义异常类注册处理器,不仅限于 HTTP 代码。在视图任何位置引发异常都会触发处理器,集中处理领域错误的响应。

flask
from werkzeug.exceptions import HTTPException

class InvalidUsage(Exception):
    status_code = 400
    def __init__(self, message, status_code=None, payload=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

@app.errorhandler(InvalidUsage)
def handle_invalid_usage(e):
    return {"error": e.message}, e.status_code

@app.route("/raise")
def raise_err():
    raise InvalidUsage("Bad input", status_code=422)

全捕获 404 处理器

通用 Exception 处理器是 500 错误的最后手段。重新引发 HTTPException 以使内置页面仍工作,并记录回溯。注意不要在生产环境中向用户泄露堆栈跟踪。

flask
@app.errorhandler(404)
def page_not_found(e):
    # Log the missing URL for analysis
    app.logger.info("404: %s", request.path)
    return render_template("errors/404.html", path=request.path), 404

@app.errorhandler(Exception)
def unhandled(e):
    # Catch any uncaught exception -> 500
    if isinstance(e, HTTPException):
        return e
    app.logger.exception("Unhandled error")
    return render_template("errors/500.html"), 500

记录错误

用 exc_info=True 记录错误以捕获完整回溯。RotatingFileHandler 防止日志文件无限增长。在生产环境中,将日志路由到集中式服务。

flask
import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5)
handler.setLevel(logging.ERROR)
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]"
))
app.logger.addHandler(handler)

@app.errorhandler(500)
def server_error(e):
    app.logger.error("Server error: %s", e, exc_info=True)
    return "Internal error", 500

HTTP 异常处理

HTTPException 是所有 Werkzeug HTTP 错误的基类。为 HTTPException 注册的处理器一次捕获所有,而子类特定处理器(NotFound)对该代码优先。

flask
from werkzeug.exceptions import HTTPException, NotFound, BadRequest

@app.errorhandler(HTTPException)
def handle_http_exception(e):
    # All HTTP exceptions in one handler
    return render_template(
        "errors/generic.html",
        code=e.code,
        name=e.name,
        description=e.description,
    ), e.code

# Specific subclasses still work
@app.errorhandler(NotFound)
def specific_404(e):
    return "custom 404", 404
13

JSON 与 API

jsonify

jsonify 将 Python 字典/列表转换为带有正确 Content-Type 和 ASCII 安全转义的 JSON 响应。Flask 2.1+ 可直接传列表;旧版本需要字典。

flask
from flask import jsonify

@app.route("/api/user")
def api_user():
    return jsonify(username="alice", age=30)

@app.route("/api/users")
def api_users():
    users = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    return jsonify(users)

# jsonify sets Content-Type: application/json
# and dumps with proper escaping

返回 JSON

直接返回字典或列表是发送 JSON 最简单的方式。自定义状态码或头部时返回元组。仅在需要完全控制序列化时使用 Response + json.dumps。

flask
# Returning a dict/list is auto-converted to JSON (Flask 1.1+)
@app.route("/api/quick")
def quick():
    return {"status": "ok", "count": 42}

# With custom status and headers
@app.route("/api/created", methods=["POST"])
def created():
    return {"id": 99}, 201, {"Location": "/api/99"}

# A Response with JSON body
from flask import Response
import json

@app.route("/api/raw")
def raw():
    return Response(json.dumps({"k": "v"}), mimetype="application/json")

接收 JSON

get_json() 在 Content-Type 为 application/json 时解析主体为 JSON。始终在使用前验证解析的结构——切勿信任客户端输入。对格式错误的有效载荷返回 400/422。

flask
from flask import request, jsonify

@app.route("/api/echo", methods=["POST"])
def echo():
    data = request.get_json()
    if not data:
        return jsonify({"error": "Invalid JSON"}), 400
    return jsonify({"received": data})

# Validate the structure
@app.route("/api/signup", methods=["POST"])
def signup():
    data = request.get_json()
    if not data or "email" not in data:
        return jsonify({"error": "email required"}), 422
    return jsonify({"email": data["email"]}), 201

API 蓝图

将 API 路由分组到带版本化 url_prefix(/api/v1)的蓝图中。这将 API 逻辑与页面路由分离,使版本控制显式化——破坏性变更时升级到 /api/v2。

flask
from flask import Blueprint, jsonify, request

api = Blueprint("api", __name__, url_prefix="/api/v1")

@api.route("/health")
def health():
    return jsonify(status="ok")

@api.route("/items", methods=["GET", "POST"])
def items():
    if request.method == "POST":
        return jsonify(created=True), 201
    return jsonify(items=[])

# Register: app.register_blueprint(api)

CORS

flask-cors 添加 Access-Control-Allow-Origin 头,使浏览器允许跨域请求。在生产环境中将来源限制为可信域名,而非使用通配符 '*'。

flask
pip install flask-cors

from flask import Flask
from flask_cors import CORS

app = Flask(__name__)

# Enable for all routes
CORS(app)

# Or restrict to specific origins
CORS(app, resources={r"/api/*": {"origins": ["https://example.com", "http://localhost:3000"]}})

# Per-route
@app.route("/api/public")
@cross_origin(origins="*")
def public():
    return {"data": "open"}

API 错误响应

辅助函数使错误响应在各端点间统一——相同的结构、相同的键名。一致的错误体使客户端错误处理更简单。包含有助于调试的详细信息,但不泄露密钥。

flask
from flask import jsonify, abort

def api_error(message, status=400, **extra):
    body = {"error": message}
    body.update(extra)
    return jsonify(body), status

@app.route("/api/users/<int:id>")
def get_user(id):
    user = find_user(id)
    if not user:
        return api_error("User not found", 404, user_id=id)
    return jsonify(user.serialize())

# Consistent error shape across the API:
# {"error": "User not found", "user_id": 1}
14

文件上传

基本上传

文件上传需要表单标签包含 enctype=multipart/form-data。设置 MAX_CONTENT_LENGTH 在主体完全读取前以 413 拒绝过大的上传——这保护服务器。

flask
from flask import request
import os

app.config["UPLOAD_FOLDER"] = "uploads"
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024  # 16 MB

@app.route("/upload", methods=["POST"])
def upload():
    file = request.files["file"]
    file.save(os.path.join(app.config["UPLOAD_FOLDER"], file.filename))
    return "uploaded"

# HTML form must include enctype:
# <form method="POST" enctype="multipart/form-data">
#   <input type="file" name="file">
#   <button>Upload</button>
# </form>

安全文件名

secure_filename 从用户提供的文件名中去除目录遍历和危险字符。它不保证唯一性——如果覆盖有风险,请追加 UUID 或检查冲突。

flask
from werkzeug.utils import secure_filename
import os

@app.route("/upload", methods=["POST"])
def upload():
    f = request.files.get("file")
    if not f:
        return "no file", 400

    safe = secure_filename(f.filename)  # "my file!!/x.txt" -> "my_file__x.txt"
    if not safe:
        return "invalid filename", 400

    f.save(os.path.join("uploads", safe))
    return f"saved as {safe}"

多文件上传

使用 request.files.getlist('files') 配合带 multiple 属性的输入可一次接受多个文件。保存前验证每个文件的扩展名和大小。

flask
from flask import request

@app.route("/upload-many", methods=["POST"])
def upload_many():
    files = request.files.getlist("files")
    saved = []
    for f in files:
        if f and f.filename:
            name = secure_filename(f.filename)
            f.save(os.path.join("uploads", name))
            saved.append(name)
    return {"saved": saved}

# <form method="POST" enctype="multipart/form-data">
#   <input type="file" name="files" multiple>
# </form>

用 WTForms 上传

Flask-WTF 的 FileField 验证上传的扩展名和存在性。表单标签须包含 enctype=multipart/form-data。上传文件可在 form.photo.data 处访问。

flask
from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileRequired, FileAllowed

class PhotoForm(FlaskForm):
    photo = FileField("Photo", validators=[
        FileRequired(),
        FileAllowed(["jpg", "jpeg", "png"], "Images only"),
    ])
    submit = SubmitField("Upload")

@app.route("/photo", methods=["GET", "POST"])
def photo():
    form = PhotoForm()
    if form.validate_on_submit():
        f = form.photo.data
        f.save(os.path.join("uploads", secure_filename(f.filename)))
        return "uploaded"
    return render_template("photo.html", form=form)

验证文件类型与大小

保存前验证扩展名和内容长度。MAX_CONTENT_LENGTH 提前拒绝超大请求,但逐文件检查提供更精细控制。为更强安全,检查文件的 magic bytes 而非仅扩展名。

flask
import os
from flask import request

ALLOWED = {".png", ".jpg", ".jpeg", ".gif", ".pdf"}
MAX_SIZE = 5 * 1024 * 1024  # 5 MB

@app.route("/upload", methods=["POST"])
def upload():
    f = request.files.get("file")
    if not f or not f.filename:
        return "no file", 400

    ext = os.path.splitext(f.filename)[1].lower()
    if ext not in ALLOWED:
        return "file type not allowed", 415

    f.stream.seek(0, 2)        # seek to end
    size = f.stream.tell()
    f.stream.seek(0)
    if size > MAX_SIZE:
        return "file too large", 413

    f.save(os.path.join("uploads", secure_filename(f.filename)))
    return "ok"
15

Cookie 与会话

设置 Cookie

通过 set_cookie 在响应对象上设置 Cookie。设 httponly=True 防止 JavaScript 读取(XSS 防御),secure=True 仅 HTTPS,samesite='Lax' 或 'Strict' 用于 CSRF 防御。

flask
from flask import make_response

@app.route("/set-cookie")
def set_cookie():
    resp = make_response("cookie set")
    resp.set_cookie("user", "alice", max_age=3600, httponly=True)
    resp.set_cookie("theme", "dark", path="/", secure=True, samesite="Lax")
    return resp

# max_age in seconds; expires sets a datetime
# httponly blocks JS access; secure requires HTTPS
# samesite: "Lax" (default), "Strict", or "None"

读取 Cookie

request.cookies 是浏览器发送的 Cookie 的只读 ImmutableMultiDict。用 .get() 带默认值而非索引,因为 Cookie 可能不存在。修改 Cookie 在响应上进行。

flask
from flask import request

@app.route("/show-cookies")
def show_cookies():
    user = request.cookies.get("user", "guest")
    theme = request.cookies.get("theme", "light")
    all_cookies = request.cookies.to_dict()
    return f"user={user}, theme={theme}"

# request.cookies is a read-only dict
# use .get() with a default to avoid KeyError

Session 对象

session 是存储序列化字典的签名 Cookie。因为用 secret_key 签名,用户无法篡改,但可以读取——因此切勿在会话中存储密码或密钥。

flask
from flask import session

app.secret_key = "a-very-secret-string"

@app.route("/login")
def login():
    session["user_id"] = 1
    session["role"] = "admin"
    return "logged in"

@app.route("/dashboard")
def dashboard():
    if "user_id" not in session:
        return "please log in", 401
    return f"user {session['user_id']}"

@app.route("/logout")
def logout():
    session.pop("user_id", None)
    return "logged out"

密钥

secret_key 签名会话 Cookie 和 CSRF 令牌。使用长随机值,并从环境变量加载以避免提交到源代码控制。所有应用实例必须共享相同密钥。

flask
import os
from flask import Flask

app = Flask(__name__)

# Generate a strong random key
app.secret_key = os.urandom(32)

# Or load from environment (recommended for production)
app.secret_key = os.environ.get("SECRET_KEY")

# For multiple workers, all must share the same key
# so sessions survive being routed to different processes.
# A key of at least 32 random bytes is recommended.

闪现消息

flash 在会话中存储消息,在重定向后仍然存活并在下次请求时显示。在模板中调用 get_flashed_messages 检索并清除。with_categories 可按类型样式化。

flask
from flask import flash, redirect, url_for, render_template

app.secret_key = "secret"   # flash uses the session

@app.route("/save", methods=["POST"])
def save():
    flash("Saved successfully!", "success")
    return redirect(url_for("index"))

# In the template:
# {% with messages = get_flashed_messages(with_categories=true) %}
#   {% for category, message in messages %}
#     <div class="alert alert-{{ category }}">{{ message }}</div>
#   {% endfor %}
# {% endwith %}

会话配置

默认情况下会话基于 Cookie,浏览器关闭时过期。设 session.permanent=True 和 PERMANENT_SESSION_LIFETIME 延长。Flask-Session 将存储移到服务端(Redis、文件系统)。

flask
app.config.update(
    SECRET_KEY="secret",
    SESSION_COOKIE_NAME="sid",
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SECURE=True,     # HTTPS only
    SESSION_COOKIE_SAMESITE="Lax",
    PERMANENT_SESSION_LIFETIME=3600,  # seconds
    SESSION_TYPE="redis",            # needs Flask-Session
)

# Mark a session as permanent
from flask import session
from datetime import timedelta

@app.route("/remember")
def remember():
    session.permanent = True
    session["user"] = "alice"
    return "remembered for 1 hour"
16

请求钩子(before/after)

before_request

before_request 处理器在每个视图前运行。用于认证检查、维护模式或将共享状态加载到 g。返回值会跳过视图并直接使用该响应。

flask
from flask import g, request

@app.before_request
def check_maintenance():
    if app.config.get("MAINTENANCE"):
        return "Site under maintenance", 503

@app.before_request
def load_user():
    token = request.headers.get("Authorization")
    if token:
        g.user = verify_token(token)

# Return a value to short-circuit the request

after_request

after_request 处理器在视图后运行,接收并返回响应。适合添加头部、日志或缓存。如果发生未处理异常,after_request 会被跳过。

flask
@app.after_request
def add_security_headers(resp):
    resp.headers["X-Content-Type-Options"] = "nosniff"
    resp.headers["X-Frame-Options"] = "SAMEORIGIN"
    return resp

@app.after_request
def log_response(resp):
    app.logger.info("%s %s -> %s", request.method, request.path, resp.status_code)
    return resp

# Must take and return a Response object

teardown_request

teardown_request 在每次请求结束时运行,即使出错——不同于 after_request。它接收异常(或 None),是释放数据库连接或文件句柄等资源的正确位置。

flask
import sqlite3
from flask import g

def get_db():
    if "db" not in g:
        g.db = sqlite3.connect("app.db")
    return g.db

@app.teardown_request
def close_db(exc):
    db = g.pop("db", None)
    if db is not None:
        db.close()

# teardown_request always runs, even if the view raised

before_first_request(已弃用)

before_first_request 在 Flask 2.3 中被移除,因为它与现代异步服务器和多进程设置不兼容。在导入时的应用上下文中或 CLI 命令中运行一次性设置。

flask
# Flask 2.3+ removed before_first_request.
# Old code:
# @app.before_first_request
# def init():
#     warm_cache()

# Modern alternative: use the app context at startup
with app.app_context():
    warm_cache()

# Or a CLI command run once during deployment
@app.cli.command("init")
def init_cmd():
    warm_cache()

修改响应

多个 after_request 处理器按注册的逆序链接。每个必须返回响应。它们用于安全头部、缓存指令和追踪 ID。

flask
@app.after_request
def no_cache(resp):
    resp.headers["Cache-Control"] = "no-store"
    resp.headers["Pragma"] = "no-cache"
    return resp

@app.after_request
def add_request_id(resp):
    rid = getattr(g, "request_id", "n/a")
    resp.headers["X-Request-ID"] = rid
    return resp

# Multiple after_request handlers run in reverse
# registration order; each receives the previous one's output.

数据库连接模式

常见模式:在 get_db 中按需打开连接(存储在 g 上),在 teardown_appcontext 中关闭。这给每个请求自己的连接,并保证即使出错也能清理。

flask
import sqlite3
from flask import g, current_app

def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(
            current_app.config["DATABASE"],
            detect_types=sqlite3.PARSE_DECLTYPES,
        )
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(exc):
    db = g.pop("db", None)
    if db is not None:
        db.close()

@app.route("/users")
def users():
    db = get_db()
    rows = db.execute("SELECT * FROM users").fetchall()
    return {"users": [dict(r) for r in rows]}
17

CLI 命令

自定义命令

用 @app.cli.command 注册 CLI 命令,用 click 参数和选项装饰。文档字符串成为帮助文本。运行 flask --help 列出所有可用命令。

flask
import click
from flask import Flask

app = Flask(__name__)

@app.cli.command("create-user")
@click.argument("name")
@click.option("--admin", is_flag=True, default=False)
def create_user(name, admin):
    """Create a new user."""
    role = "admin" if admin else "user"
    click.echo(f"Created {name} as {role}")

# Run: flask create-user alice --admin

flask run

flask run 启动 Werkzeug 开发服务器。--debug 启用重载器和交互式调试器。--app 让你指向模块或工厂。切勿在生产环境使用此服务器——它是单线程且不安全的。

flask
# Basic
flask run

# Custom host/port
flask run --host=0.0.0.0 --port=8000

# Enable reloader and debugger
flask run --debug

# Specify the app explicitly
flask --app myapp run
flask --app myapp:create_app run

# With SSL (development)
flask run --cert=cert.pem --key=key.pem

Shell 上下文

shell_context_processor 向 flask shell 注入名称,无需导入即可实验。这对调试模型和对应用运行临时查询非常宝贵。

flask
from flask import Flask
from models import db, User

app = Flask(__name__)

@app.shell_context_processor
def make_shell_context():
    return {"db": db, "User": User, "app": app}

# flask shell
# >>> User.query.all()
# >>> db.session.add(User(username="x"))

带 CLI 的应用工厂

使用工厂模式时,在 create_app 内注册 CLI 命令,使其能访问配置好的应用。Flask 的 --app 选项检测工厂并自动调用它。

flask
# myapp/__init__.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    app.config.from_object("config.Config")

    @app.cli.command("init-db")
    def init_db():
        """Initialize the database."""
        click.echo("Initialized")

    return app

# Run: flask --app myapp init-db
# The factory is detected automatically.

CLI 分组

用 @app.cli.group 将相关命令分组到父级下。子命令注册在组上,保持 CLI 有组织:flask db init、flask db migrate 等。

flask
import click
from flask import Flask

app = Flask(__name__)

@app.cli.group("db")
def db_group():
    """Database commands."""
    pass

@db_group.command("init")
def db_init():
    click.echo("DB initialized")

@db_group.command("drop")
def db_drop():
    click.echo("DB dropped")

# flask db init
# flask db drop

环境变量

FLASK_APP 和 FLASK_DEBUG 配置 CLI。用 .flaskenv 存放共享开发设置,.env 存放密钥(安装 python-dotenv 后自动加载)。密钥不要放在 .flaskenv 中,因为通常会被提交。

flask
# Flask reads these env vars:
# FLASK_APP      -> module/app to run (app.py, myapp, myapp:create_app)
# FLASK_DEBUG    -> 1/0 to toggle debug mode
# FLASK_ENV      -> development/production (deprecated, use FLASK_DEBUG)
# FLASK_RUN_PORT -> default port
# FLASK_RUN_HOST -> default host

# .env or .flaskenv files are auto-loaded by python-dotenv
# .flaskenv (committed): FLASK_APP=app.py
# .env (gitignored): SECRET_KEY=..., DATABASE_URL=...

# Prefer app.config.from_envvar or from_object for app secrets.
18

测试

Test Client 基础

test_client() 返回一个无需网络即可发起请求的客户端。设 TESTING=True 以获得更好的错误消息并禁用错误捕获。使用 fixture 让每个测试获得新应用。

flask
import pytest
from myapp import create_app

@pytest.fixture()
def client():
    app = create_app()
    app.config["TESTING"] = True
    with app.test_client() as client:
        with app.app_context():
            pass  # init DB here
        yield client

def test_home(client):
    resp = client.get("/")
    assert resp.status_code == 200
    assert b"Hello" in resp.data

在测试中发起请求

测试客户端模拟浏览器:get/post/put/delete 带有 data(表单)、json、query_string 和 headers。用 resp.get_json() 解析 JSON 响应,resp.headers 检查头部。

flask
def test_get(client):
    resp = client.get("/users/1")
    assert resp.status_code == 200

def test_post_form(client):
    resp = client.post("/login", data={"username": "a", "password": "b"})
    assert resp.status_code == 302  # redirect

def test_post_json(client):
    resp = client.post("/api/echo", json={"key": "value"})
    assert resp.get_json() == {"received": {"key": "value"}}

def test_query(client):
    resp = client.get("/search?q=flask&page=2")
    assert resp.status_code == 200

def test_headers(client):
    resp = client.get("/", headers={"X-Test": "1"})

测试 JSON API

传入 json=... 发送带正确 Content-Type 的 JSON 主体。断言状态、解析的 JSON 主体和 Location 等头部。测试成功和错误路径确保 API 行为一致。

flask
def test_create_user(client):
    resp = client.post(
        "/api/users",
        json={"username": "alice", "email": "[email protected]"},
    )
    assert resp.status_code == 201
    body = resp.get_json()
    assert body["username"] == "alice"
    assert "id" in body
    assert resp.headers["Location"].endswith(str(body["id"]))

def test_error(client):
    resp = client.post("/api/users", json={})
    assert resp.status_code == 422
    assert "error" in resp.get_json()

pytest Fixtures

Fixture 提供可复用的设置:内存数据库、测试客户端和 CLI 运行器。yield 资源并在之后清理。test_cli_runner() 在测试中调用 CLI 命令并检查输出和退出码。

flask
import pytest
from myapp import create_app, db

@pytest.fixture()
def app():
    app = create_app()
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()

@pytest.fixture()
def client(app):
    return app.test_client()

@pytest.fixture()
def runner(app):
    return app.test_cli_runner()

def test_cli(runner):
    result = runner.invoke(args=["create-user", "alice"])
    assert result.exit_code == 0

测试数据库

使用内存 SQLite 数据库进行快速、隔离的测试。在 fixture 内创建表和种子数据,之后删除所有内容。每个测试从干净状态开始,防止测试间污染。

flask
import pytest
from myapp import create_app, db
from myapp.models import User

@pytest.fixture()
def app():
    app = create_app()
    app.config["TESTING"] = True
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
    with app.app_context():
        db.create_all()
        db.session.add(User(username="alice"))
        db.session.commit()
        yield app
        db.session.remove()
        db.drop_all()

def test_user_count(app):
    with app.app_context():
        assert User.query.count() == 1

Mocking

patch 用 mock 替换外部调用(HTTP、邮件、队列),使测试保持快速和确定性。断言 mock 以预期参数被调用。始终 patch 名称查找的位置,而非定义的位置。

flask
from unittest.mock import patch, MagicMock
import myapp

def test_external_call(client):
    with patch("myapp.requests.get") as mock_get:
        mock_get.return_value = MagicMock(status_code=200, json=lambda: {"ok": True})
        resp = client.get("/proxy")
        assert resp.status_code == 200
        mock_get.assert_called_once()

def test_email(client):
    with patch("myapp.send_email") as mock_send:
        client.post("/register", data={"email": "[email protected]"})
        mock_send.assert_called_once_with("[email protected]", "Welcome!")
19

部署

生产服务器警告

Werkzeug 开发服务器是单线程且不安全的——切勿暴露到互联网。使用生产 WSGI 服务器(Gunicorn、uWSGI、Waitress)在反向代理之后。debug 调试器是远程代码执行风险。

flask
# WARNING: Flask's built-in server is for development only!
# app.run() / flask run -> single-threaded, no security, not scalable

# For production use a WSGI server:
# - Gunicorn (Linux/macOS)
# - uWSGI
# - Waitress (Windows-friendly)
# Behind a reverse proxy (Nginx, Apache, Caddy)

# The dev server also enables the interactive debugger
# when debug=True, which allows arbitrary code execution.
# NEVER deploy with debug=True.

Gunicorn

Gunicorn 是 Linux/macOS 上最流行的 WSGI 服务器。公式 (2 x CPU + 1) 个 worker 是良好的起点。在 Nginx 后绑定 127.0.0.1,让代理处理 TLS 和静态文件。

flask
pip install gunicorn

# Basic
gunicorn "app:app"

# With factory
gunicorn "myapp:create_app()"

# Workers and binding
gunicorn -w 4 -b 0.0.0.0:8000 "app:app"

# With timeout and logging
gunicorn -w 4 -b 0.0.0.0:8000 --timeout 120 --access-logfile - "app:app"

# Recommended workers: (2 * CPU) + 1
# Use --preload to share memory, or -k gevent for async.

uWSGI

uWSGI 是高性能、高度可配置的 WSGI 服务器。master 进程管理 worker;vacuum 在退出时清理。与 Nginx 配合使用 uwsgi 协议获得最佳吞吐量。

flask
pip install uwsgi

# Command line
uwsgi --http 0.0.0.0:8000 --module app:app --processes 4 --threads 2

# INI config (uwsgi.ini)
# [uwsgi]
# module = app:app
# http = 0.0.0.0:8000
# processes = 4
# threads = 2
# master = true
# vacuum = true
# die-on-term = true

# Run: uwsgi uwsgi.ini

反向代理(Nginx)

Nginx 位于 Gunicorn 前面:直接提供静态文件(更快)、终止 TLS、将请求转发到 WSGI 服务器。X-Forwarded-* 头让 Flask 看到真实客户端 IP 和协议——设 ProxyFix 以信任它们。

flask
# /etc/nginx/sites-available/myapp
server {
    listen 80;
    server_name example.com;

    location /static/ {
        alias /path/to/app/static/;
        expires 30d;
    }

    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;
        proxy_set_header X-Forwarded-Proto \$scheme;
    }
}

Docker

精简的 Python 基础镜像保持镜像小巧。在复制代码前安装依赖以利用 Docker 的层缓存。通过环境变量(--env-file)传递密钥,切勿烘焙到镜像中。

flask
# Dockerfile
FROM python:3.12-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:app"]

# .dockerignore
# __pycache__/
# venv/
# *.pyc
# .env

# Build and run
# docker build -t myapp .
# docker run -p 8000:8000 --env-file .env myapp

环境配置

分层配置:代码中的默认值、环境特定的覆盖、来自环境变量的实例密钥。这使密钥不进入源代码控制,让一套代码在开发、预发布和生产中运行。

flask
import os
from flask import Flask

app = Flask(__name__)
app.config.from_object("config.DefaultConfig")

# Override with environment-specific config
env = os.environ.get("FLASK_ENV", "development")
app.config.from_object(f"config.{env.title()}Config")

# Then instance-specific secrets from env vars
app.config["SECRET_KEY"] = os.environ["SECRET_KEY"]
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ["DATABASE_URL"]

# Use python-dotenv for local .env files
from dotenv import load_dotenv
load_dotenv()
20

配置

配置对象

将配置定义为继承基础 Config 的类。仅覆盖每个环境不同的内容。from_object 将大写属性复制到 app.config,小写辅助函数被忽略。

flask
class Config:
    DEBUG = False
    TESTING = False
    SECRET_KEY = "change-me"
    SQLALCHEMY_DATABASE_URI = "sqlite:///app.db"

class DevelopmentConfig(Config):
    DEBUG = True
    SECRET_KEY = "dev-secret"

class ProductionConfig(Config):
    DEBUG = False
    SECRET_KEY = os.environ["SECRET_KEY"]

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"

app.config.from_object(DevelopmentConfig)

环境变量

使用 os.environ 存放密钥,使其永不在代码中。from_envvar 加载环境变量指向的 Python 文件——用于部署特定覆盖。python-dotenv 为本地开发加载 .env 文件。

flask
import os
from flask import Flask

app = Flask(__name__)

# Read individual values
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev")
app.config["DATABASE_URL"] = os.environ.get("DATABASE_URL", "sqlite:///app.db")

# Load a whole config file from an env var path
# export MYAPP_SETTINGS=/etc/myapp/prod.py
app.config.from_envvar("MYAPP_SETTINGS", silent=True)

# python-dotenv loads .env automatically if installed
from dotenv import load_dotenv
load_dotenv()

Instance 文件夹

instance 文件夹在版本控制之外存放部署特定文件(密钥、配置)。设 instance_relative_config=True 并用 from_pyfile 加载。这干净地将代码与环境配置分离。

flask
# The instance/ folder sits outside the package, not in VCS.
# Flask searches it automatically for config files.

app = Flask(__name__, instance_relative_config=True)

# Load from instance/config.py
app.config.from_pyfile("config.py", silent=True)

# instance/config.py (deployment-specific, gitignored)
# SECRET_KEY = "real-production-key"
# SQLALCHEMY_DATABASE_URI = "postgresql://user:pass@db/app"

# Access the path
print(app.instance_path)  # /abs/path/to/instance

从文件加载配置

from_object 导入模块或对象路径;from_pyfile 加载绝对文件;from_file(Flask 2.0+)使用加载器函数,支持 JSON、TOML 或 YAML。选择一种格式并保持一致。

flask
# config.py
DEBUG = True
SECRET_KEY = "dev"
SQLALCHEMY_DATABASE_URI = "sqlite:///dev.db"

# Load a Python file
app.config.from_object("config")

# Load by path
app.config.from_pyfile("/etc/myapp/config.py")

# Load from a JSON file
app.config.from_file("config.json", load=json.load)

# config.json
# { "DEBUG": true, "SECRET_KEY": "dev" }

开发 vs 生产

通过环境变量切换配置,使相同代码在各处运行。生产必须禁用 debug(调试器是安全漏洞)并使用真实数据库。如果生产中开启了 debug 则记录警告。

flask
import os
from flask import Flask

env = os.environ.get("FLASK_ENV", "development")

configs = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
    "testing": TestingConfig,
}

app = Flask(__name__)
app.config.from_object(configs[env])

# In production, never enable debug:
if app.config["DEBUG"]:
    print("WARNING: running in debug mode!")

日志配置

仅在非 debug 模式下配置日志(Flask 在 debug 中默认输出到 stderr)。RotatingFileHandler 防止日志无限增长。生产中考虑为聚合服务使用结构化 JSON 日志。

flask
import logging
from logging.handlers import RotatingFileHandler

if not app.debug:
    handler = RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5)
    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)

# In a view
@app.route("/")
def index():
    app.logger.info("Index page accessed")
    return "Hello"

# Flask's logger is a standard logging.Logger

这篇内容对您有帮助吗?

学习路径

从零开始学习

通过结构化课程从头学习这个语言。