入门
第一个应用
Flask 是一个微框架——它提供基础功能,让你按需添加。debug=True 启用自动重载和详细错误页面。
# 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 使其在网络中可访问。
# 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 复现环境。
# 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/ 文件夹存放不应纳入版本控制的部署特定密钥。
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 选项接受返回应用的回调函数。
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 知道去哪里找静态文件和模板。如果布局与默认不同,可以用构造函数参数覆盖文件夹位置。
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!"路由
变量规则
用 <converter:varname> 标记的部分捕获 URL 片段。内置转换器:string(默认)、int、float、path(接受斜杠)、uuid。值作为函数参数传入。
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 子类可强制正则模式,如产品代码。
@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。每个路由选择一种约定并保持一致。
@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 FoundHTTP 方法
使用 methods 参数接受多种动词,或用 @app.get / @app.post 快捷方式(Flask 2.0+)处理单方法路由。GET 是默认方法。
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。
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 形式。
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.请求与响应
请求对象
request 是绑定到当前请求上下文的线程局部代理。它暴露 URL、方法、头部、主体和客户端信息,无需显式传参。
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) 元组。
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 可强制浏览器下载。
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 状态码表达语义。
@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。
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 是响应的一部分,不是请求主体。
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请求对象(form/args/files)
查询参数(args)
request.args 是查询字符串值的 MultiDict。使用 get 带默认值和类型转换,用 getlist 处理重复键。切勿直接索引——缺失键会引发 400。
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 为空。
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 处理用户提供的文件名,以去除路径分隔符和危险字符。
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 抑制解析错误。
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。
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 是 '?' 后的原始字节。
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(),
}Jinja2 模板
渲染模板
render_template 从 templates/ 文件夹加载 Jinja2 文件,并将关键字参数注入为变量。{{ }} 语法输出表达式;Flask 自动转义 HTML 以保证安全。
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 过滤器将字符串标记为可信以避免转义——仅用于你完全控制的内容。
# 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 在可迭代对象为空时执行。
{# 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() 渲染父块的内容。
{# 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 上注册自定义过滤器可在模板间复用逻辑。
{{ 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。
{# 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>静态文件
提供静态文件
Flask 自动为 static 文件夹注册 /static/<path:filename> 路由。始终用 url_for 构建 URL,即使 static_url_path 改变也能保持路径正确。
# 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(用于邮件或订阅源),传入查询参数用于缓存清除。
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。
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.cssFavicon
浏览器默认请求根路径的 /favicon.ico。添加显式路由或通过 <link> 标签链接图 标。send_from_directory 安全地从目录提供文件。
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)可在大型项目中自动化此过程。
# 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 }}">蓝图
创建蓝图
蓝图将相关路由、模板和静态文件分组到一个模块。第一个参数是蓝图名称;url_prefix 为其定义的每个路由前置一个路径。
# 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,使同一蓝图在不同应用中挂载到不同路径。
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 而不破坏现有客户端。
# 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...蓝图资源
蓝图可携带自己的模板和静态文件夹。模板解析先检查蓝图,再检查应用,因此蓝图本地文件可覆盖同名的应用级文件。
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')蓝图错误处理器
蓝图级错误处理器仅捕获该蓝图路由引发的错误,提供特定分区的错误页面。应用级处理器处理未捕获的内容。
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 会拼接,让你用小模块组合大型应用。
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上下文(g/session)
应用上下文
应用上下文使 current_app 和 g 在请求外可用(如 CLI 命令或后台任务中)。在请求外工作时用 app.app_context() 推入上下文。
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 处理器中清理。
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 服务器。
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 对象,将属性访问转发到活动上下文。切勿将它们捕获到全局变量中——始终在调用时访问。
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 对其签名。值存储在客户端且用户可读取(但不可篡改),因此切勿存储密钥。
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_* 处理器始终运行(即使在出错时),因此是关闭数据库连接和释放资源的正确位置。
# 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)SQLAlchemy 集成
安装 Flask-SQLAlchemy
Flask-SQLAlchemy 封装 SQLAlchemy 并与 Flask 的应用上下文集成。设置 SQLALCHEMY_DATABASE_URI 为连接字符串,禁用 track modifications 以节省内 存。
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 添加约束。
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() 返回条目和分页元数据。
# 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() 回滚以保持会话可用。
# 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' 返回查询对象而非列表,适合需要进一步过滤的大型集合。
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 应用。始终在提交前审查自动生成的脚本。
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 revisionWTForms(Flask-WTF)
安装 Flask-WTF
Flask-WTF 将 WTForms 与 Flask 集成,并自动添加 CSRF 保护。需要 SECRET_KEY 以便签名 CSRF 令牌。仅在公共 API 端点禁用 CSRF。
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() 时运行的验证器列表。验证器强制必填、长度、邮箱格式等。
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 输入。
{# 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 读取值。失败时重新渲染模板显示错误。
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 常用于密码确认。在列表中链式调用,按顺序运行直到一个失败。
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。
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)Flask-Login 认证
安装 Flask-Login
Flask-Login 管理用户会话:登录、登出和记住用户。设置 login_view 以便未认证用户重定向到那里。login_message 在重定向时闪现。
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 返回用户。切勿存储明文密码——必须哈希。
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。登录前始终通过哈希检查验证密码,失败时闪现通用错误。
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)的代理,在视图和模板中可用。
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)。
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。
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 %}错误处理
abort()
abort() 立即停止视图并返回 HTTP 错误。传入状态码和可选消息。它引发 HTTPException,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() 调用和映射到该状态的异常。
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 代码。在视图任何位置引发异常都会触发处理器,集中处理领域错误的响应。
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 以使内置页面仍工作,并记录回溯。注意不要在生产环境中向用户泄露堆栈跟踪。
@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 防止日志文件无限增长。在生产环境中,将日志路由到集中式服务。
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", 500HTTP 异常处理
HTTPException 是所有 Werkzeug HTTP 错误的基类。为 HTTPException 注册的处理器一次捕获所有,而子类特定处理器(NotFound)对该代码优先。
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", 404JSON 与 API
jsonify
jsonify 将 Python 字典/列表转换为带有正确 Content-Type 和 ASCII 安全转义的 JSON 响应。Flask 2.1+ 可直接传列表;旧版本需要字典。
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。
# 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。
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"]}), 201API 蓝图
将 API 路由分组到带版本化 url_prefix(/api/v1)的蓝图中。这将 API 逻辑与页面路由分离,使版本控制显式化——破坏性变更时升级到 /api/v2。
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 头,使浏览器允许跨域请求。在生产环境中将来源限制为可信域名,而非使用通配符 '*'。
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 错误响应
辅助函数使错误响应在各端点间统一——相同的结构、相同的键名。一致的错误体使客户端错误处理更简单。包含有助于调试的详细信息,但不泄露密钥。
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}文件上传
基本上传
文件上传需要表单标签包含 enctype=multipart/form-data。设置 MAX_CONTENT_LENGTH 在主体完全读取前以 413 拒绝过大的上传——这保护服务器。
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 或检查冲突。
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 属性的输入可一次接受多个文件。保存前验证每个文件的扩展名和大小。
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 处访问。
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 而非仅扩展名。
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"请求钩子(before/after)
before_request
before_request 处理器在每个视图前运行。用于认证检查、维护模式或将共享状态加载到 g。返回值会跳过视图并直接使用该响应。
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 requestafter_request
after_request 处理器在视图后运行,接收并返回响应。适合添加头部、日志或缓存。如果发生未处理异常,after_request 会被跳过。
@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 objectteardown_request
teardown_request 在每次请求结束时运行,即使出错——不同于 after_request。它接收异常(或 None),是释放数据库连接或文件句柄等资源的正确位置。
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 raisedbefore_first_request(已弃用)
before_first_request 在 Flask 2.3 中被移除,因为它与现代异步服务器和多进程设置不兼容。在导入时的应用上下文中或 CLI 命令中运行一次性设置。
# 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。
@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 中关闭。这给每个请求自己的连接,并保证即使出错也能清理。
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]}CLI 命令
自定义命令
用 @app.cli.command 注册 CLI 命令,用 click 参数和选项装饰。文档字符串成为帮助文本。运行 flask --help 列出所有可用命令。
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 --adminflask run
flask run 启动 Werkzeug 开发服务器。--debug 启用重载器和交互式调试器。--app 让你指向模块或工厂。切勿在生产环境使用此服务器——它是单线程且不安全的。
# 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.pemShell 上下文
shell_context_processor 向 flask shell 注入名称,无需导入即可实验。这对调试模型和对应用运行临时查询非常宝贵。
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 选项检测工厂并自动调用它。
# 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 等。
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 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.测试
Test Client 基础
test_client() 返回一个无需网络即可发起请求的客户端。设 TESTING=True 以获得更好的错误消息并禁用错误捕获。使用 fixture 让每个测试获得新应用。
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 检查头部。
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 行为一致。
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 命令并检查输出和退出码。
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 内创建表和种子数据,之后删除所有内容。每个测试从干净状态开始,防止测试间污染。
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() == 1Mocking
patch 用 mock 替换外部调用(HTTP、邮件、队列),使测试保持快 速和确定性。断言 mock 以预期参数被调用。始终 patch 名称查找的位置,而非定义的位置。
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!")部署
生产服务器警告
Werkzeug 开发服务器是单线程且不安全的——切勿暴露到互联网。使用生产 WSGI 服务器(Gunicorn、uWSGI、Waitress)在反向代理之后。debug 调试器是远程代码执行风险。
# 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 和静态文件。
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 协议获得最佳吞吐量。
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 以信任它们。
# /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)传递密钥,切勿烘焙到镜像中。
# 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环境配置
分层配置:代码中的默认值、环境特定的覆盖、来自环境变量的实例密钥。这使密钥不进入源代码控制,让一套代码在开发、预发布和生产中运行。
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()配置
配置对象
将配置定义为继承基础 Config 的类。仅覆盖每个环境不同的内容。from_object 将大写属性复制到 app.config,小写辅助函数被忽略。
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 文件。
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 加载。这干净地将代码与环境配置分离。
# 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。选择一种格式并保持一致。
# 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 则记录警告。
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 日志。
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相关 Flask 代码片段
Copy-paste ready code for common tasks.
Routing
Define routes with methods and dynamic parameters.
Request and Response
Access request data and build custom responses.
Jinja2 Templates
Render templates with context and template inheritance.
Blueprints
Organize an app into modular blueprints.
Session
Store per-user data in signed session cookies.
Error Handling
Register custom error handlers for HTTP exceptions.
Flask-SQLAlchemy
Define models and query them with Flask-SQLAlchemy.
Custom Decorators
Build decorators to gate or augment view functions.
这篇内容对您有帮助吗?