高级篇 / 第 18 章

第 18 章 · Web 开发入门（Flask）

Web 开发全景：一个请求的一生

当你在浏览器输入 http://127.0.0.1:5000/hello 并回车，背后发生了一连串动作：

1. DNS 解析：把域名翻译成 IP 地址（本机就是 127.0.0.1，直接跳过）。
2. TCP 连接：客户端与服务器的 80/443 端口建立连接（HTTP/1.1、H/2）。
3. TLS 握手：HTTPS 才需要，协商证书与加密密钥。
4. 发送 HTTP 请求：请求行 + 请求头 + 请求体。
5. 服务器处理：路由匹配 → 调用视图函数 → 拿到返回值。
6. 返回 HTTP 响应：状态行 + 响应头 + 响应体，浏览器渲染。

整套 Web 框架（Python、Java、PHP…）本质上只做三件事：解析请求、调用你的代码、封装响应。Flask 把这三件事做到了极致的简洁：「它是一个 WSGI 应用，只把 Flask 给你。」

HTTP 协议：Web 的共同语言

HTTP（HyperText Transfer Protocol，超文本传输协议）是一种无状态（stateless）的请求-响应协议：客户端发请求，服务器回响应，每次请求彼此独立，服务器不记得你上次来过（无状态由 Cookie/Session 机制补上）。

请求报文结构

POST /api/users?page=1 HTTP/1.1        <-- 请求行：方法 + 路径 + 协议版本
Host: example.com                      <-- 请求头：键值对
Content-Type: application/json
Content-Length: 27
User-Agent: Mozilla/5.0
Cookie: session=abc123
Authorization: Bearer eyJ...
                                        <-- 空行分隔
{"name": "alice", "age": 28}           <-- 请求体

响应报文结构

HTTP/1.1 200 OK                       <-- 状态行：协议版本 + 状态码 + 状态短语
Content-Type: application/json
Content-Length: 28
Set-Cookie: session=xyz789; HttpOnly
                                        <-- 空行
{"id": 1, "name": "alice"}

常用请求方法

状态码速查

无状态与会话

HTTP 本身不记住你是谁，所以出现了 Cookie + Session 机制：服务器把一段随机 ID 放进 Cookie 发给浏览器，浏览器每次请求自动带上，服务器据此找回服务端保存的会话数据。

Set-Cookie: session=eyJpZCI6MTB9.abc123; HttpOnly; Secure; SameSite=Lax; Max-Age=3600

# Cookie 属性
# HttpOnly  禁止 JS 读取 -> 缓解 XSS 窃取
# Secure    只在 HTTPS 传输
# SameSite  Lax/Strict  跨站请求防护 -> 缓解 CSRF

WSGI：Python Web 的标准接口

WSGI（Web Server Gateway Interface，Web 服务器网关接口）是 Python 官方的 Web 服务器与 Web 应用之间的接口规范，定义于 PEP 3333。它解决了两个问题：服务器该调用哪个函数、返回值和异常该怎么传。

可调用对象（application）

# 一个符合 WSGI 规范的最小应用
def app(environ, start_response):
    """
    environ: dict，包含请求信息（类 CGI 变量 + wsgi.* 扩展键）
    start_response: 可调用，用于设置状态行和响应头
    返回值: 一个可迭代对象（通常产出 bytes）
    """
    start_response("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
    return [f"你好，{environ['PATH_INFO']}！".encode("utf-8")]


# 启动开发服务器（werkzeug 提供）
from werkzeug.serving import run_simple
run_simple("127.0.0.1", 5000, app, with_debugger=True)

environ 里有什么

def app(environ, start_response):
    if environ["REQUEST_METHOD"] == "POST":
        try:
            length = int(environ.get("CONTENT_LENGTH") or 0)
            body = environ["wsgi.input"].read(length)      # 必须按长度读，不能 read() 到 EOF
        except (ValueError, OSError):
            body = b""
    else:
        body = b""

    # start_response 必须在返回 body 之前调用
    start_response("200 OK", [("Content-Type", "application/json; charset=utf-8")])
    import json
    return [json.dumps({
        "method": environ["REQUEST_METHOD"],
        "path": environ["PATH_INFO"],
        "body": body.decode("utf-8", "replace"),
    }, ensure_ascii=False).encode("utf-8")]

Flask 快速上手

Flask 是 Werkzeug（WSGI 工具库）和 Jinja2（模板引擎）的上层封装，特点是「微型但可扩展」：核心极简，功能靠扩展（extension）按需添加。

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install flask

app.py —— 第一个 Flask 应用

from flask import Flask

app = Flask(__name__)

@app.route("/")
def index() -> str:
    return "欢迎来到 Flask！"

@app.route("/hello/<name>")
def hello(name: str) -> str:
    return f"你好，{name}！"

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=True)

运行 python app.py，浏览器打开 http://127.0.0.1:5000/。访问 /hello/张三 会看到「你好，张三！」。

工厂模式：推荐的工程结构

myapp/
├── app.py              # create_app 工厂函数
├── config.py           # 配置
├── extensions.py       # 扩展实例
├── blueprints/
│   ├── __init__.py
│   ├── main.py
│   └── api.py
├── templates/
├── static/
└── run.py

# extensions.py —— 扩展在模块级创建，在工厂里 init_app 绑定
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

# config.py —— 按环境分配置
import os

class BaseConfig:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-change-me")
    JSON_AS_ASCII = False


class DevelopmentConfig(BaseConfig):
    DEBUG = True


class ProductionConfig(BaseConfig):
    DEBUG = False
    SECRET_KEY = os.environ["SECRET_KEY"]   # 生产环境必须从环境变量读

    @classmethod
    def check(cls) -> None:
        if cls.SECRET_KEY.startswith("dev-"):
            raise RuntimeError("生产环境不能使用默认 SECRET_KEY")


config = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
}[os.environ.get("FLASK_ENV", "development")]

# app.py —— 应用工厂
from flask import Flask, redirect, url_for

from config import config
from extensions import db


def create_app(config_name: str | None = None) -> Flask:
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_object(config if config_name is None else config[config_name])

    db.init_app(app)                       # 初始化扩展

    from blueprints.main import bp as main_bp
    from blueprints.api import bp as api_bp
    app.register_blueprint(main_bp)
    app.register_blueprint(api_bp, url_prefix="/api", version="1")   # -> /api/v1/...

    # 上下文处理器：注入所有模板都能用的变量
    @app.context_processor
    def inject_globals() -> dict:
        return {"app_name": "MyApp", "version": "1.0.0"}

    return app

# run.py —— 启动入口
from app import create_app

app = create_app()

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

路由：URL 与函数的映射

路由（routing）就是把 URL 规则映射到处理函数。Flask 用 werkzeug.routing 实现，内部维护一个 URL 适配器（URLAdapter），支持静态段、动态段、类型转换和优先级排序。

各类路由写法

from flask import Flask, request, redirect, url_for, abort

app = Flask(__name__)

# 静态路由
@app.route("/about")
def about():
    return "关于页"


# 动态段 + 类型转换器
@app.route("/user/<int:uid>")                       # int 只匹配数字，负数要写 <int:uid>
def user_detail(uid: int):
    return f"用户 {uid}"

@app.route("/post/<float:score>")
def post_by_score(score: float):
    return f"分数 {score}"

@app.route("/post/<path:subpath>")                    # path 匹配含斜杠的剩余路径
def post_path(subpath: str):
    return f"子路径：{subpath}"


# 多规则 + 多端点
@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        return "处理登录"
    return "登录页面"


# 默认值：地址是 /page 或 /page/ 都匹配
@app.route("/page", defaults={"num": 1})
@app.route("/page/<int:num>")
def page(num: int):
    return f"第 {num} 页"


# 唯一端点名，避免视图函数名冲突
@app.route("/old", endpoint="legacy")
def legacy_view():
    return "老接口"

常用类型转换器

# 自定义转换器
from werkzeug.routing import BaseConverter

class ListConverter(BaseConverter):
    """把逗号分隔的字符串转成 list。"""
    regex = r"[\d,]+"

    def to_python(self, value: str) -> list[int]:
        return [int(x) for x in value.split(",")]

    def to_url(self, value: list[int]) -> str:
        return ",".join(str(x) for x in value)


app.url_map.converters["list"] = ListConverter

@app.route("/ids/<list:ids>")
def by_ids(ids: list[int]):
    return f"收到 ID 列表：{ids}"     # 访问 /ids/1,2,3 -> [1, 2, 3]

请求对象 request

@app.route("/echo", methods=["GET", "POST"])
def echo():
    # 请求头
    ua = request.headers.get("User-Agent")
    # 查询参数
    kw = request.args.get("kw", "")            # ?kw=flask
    page = request.args.get("page", default=1, type=int)
    all_args = request.args.to_dict(flat=False)   # 多值 {'tag': ['a','b']}
    # 请求体
    if request.is_json:
        data = request.get_json()               # 解析 application/json
    else:
        data = request.form                      # 表单字典（MultiDict）
    raw = request.data                           # 原始 bytes
    # 文件上传
    f = request.files.get("avatar")
    # 请求上下文
    ip = request.remote_addr
    scheme = request.scheme
    path = request.path
    full = request.url
    if request.is_json:
        return data
    return {"ua": ua, "kw": kw, "page": page, "form": dict(request.form),
            "ip": ip, "scheme": scheme, "path": path, "url": full}

url_for：反向生成 URL

from flask import url_for, redirect, request

@app.route("/profile/<int:uid>")
def profile(uid: int):
    return f"页面：{uid}"

@app.route("/go")
def go():
    # ✅ 推荐：硬编码路径以后改 URL 只需改一处
    return redirect(url_for("profile", uid=42))       # -> /profile/42
    # 带查询参数
    # return redirect(url_for("profile", uid=42, tab="posts"))

@app.route("/search")
def search():
    kw = request.args.get("kw")
    if not kw:
        # 把参数原样带回列表页
        return redirect(url_for("index", kw=kw, page=2))
    return f"搜索：{kw}"

模板引擎 Jinja2

Jinja2 支持 {{ 表达式 }} 输出、{% 语句 %} 控制流。它的一个核心安全特性是自动转义（autoescape）：.html 模板中的变量会被自动 HTML 转义，这是防 XSS 的第一道防线。

templates/index.html

<!DOCTYPE html>
<html lang="zh-CN">
<head><title>{{ title }}</title></head>
<body>
  <h1>{{ title }}</h1>

  <!-- 变量与表达式 -->
  <p>欢迎，{{ user.name }}，你有 {{ user.messages | length }} 条消息。</p>
  <p>今天是 {{ now.strftime('%Y-%m-%d') }}</p>
  <p>{{ 10 / 3 }} -> 3.3333333333333335（除法返回浮点）</p>

  <!-- 条件 -->
  {% if user.vip %}
    <p>VIP 会员</p>
  {% elif user.points > 100 %}
    <p>积分充足</p>
  {% else %}
    <p>普通用户</p>
  {% endif %}

  <!-- 循环 -->
  <ul>
  {% for item in items %}
    <li>{{ loop.index }}. {{ item.name }} — ￥{{ item.price }}{% if item.sale %} <em>促销</em>{% endif %}</li>
  {% else %}
    <li>暂无数据</li>
  {% endfor %}
  </ul>

  <!-- loop 变量速查 -->
  <p>items 总数：{{ items|length }}</p>

  <!-- 过滤器和运算 -->
  <p>{{ '  hello  ' | trim | upper }}</p>
  <p>{{ [3,1,2] | sort }}</p>
  <p>{{ 'abc' | length }}</p>
  <p>{{ price | round(2) }}</p>
  <p>{{ text | default('无') }}</p>
  <p>{{ '<b>hi</b>' }}</p>              <!-- 自动转义，输出 &lt;b&gt;hi&lt;/b&gt; -->
</body>
</html>

常用过滤器（filter）对照

渲染模板

from datetime import datetime
from flask import render_template, abort

@app.route("/")
def index():
    items = [
        {"name": "机械键盘", "price": 399.0, "sale": True},
        {"name": "人体工学椅", "price": 1299.0, "sale": False},
    ]
    # 变量名可以直接传，也可以传字典：render_template("index.html", title="首页", items=items)
    return render_template(
        "index.html",
        title="商品列表",
        user={"name": "张三", "vip": True, "points": 200, "messages": [1, 2, 3]},
        items=items,
        now=datetime.now(),
        price=3.14159,
        text="",
        html="<b>hi</b>",
        long_text="一" * 200,
    )


@app.route("/user/<int:uid>")
def user_detail(uid: int):
    user = get_user(uid)          # 自己的数据库查询
    if user is None:
        abort(404)                 # 抛 404，走错误处理器
    return render_template("user.html", user=user)


# 模板复用：base.html + 继承
# templates/base.html
# <body>
#   {% block content %}默认内容{% endblock %}
# </body>
# templates/page.html
# {% extends "base.html" %}
# {% block content %}
#   <h1>本页内容</h1>
# {% endblock %}

# 模板中引用静态文件：{{ url_for('static', filename='style.css') }}

include 与 import

<div class="card">
  <h3>{{ item.title }}</h3>
  <p>{{ item.desc }}</p>
</div>


{% for item in items %}
  {% include "_card.html" %}
{% endfor %}


{% import "macros.html" as ui %}
{{ ui.render_button("提交", "primary") }}


{% from "macros.html" import render_button with context %}


{% macro render_button(text, style="default") -%}
  <button class="btn btn-{{ style }}">{{ text }}</button>
{%- endmacro %}

静态文件与表单处理

静态文件

放在 static/ 目录下的文件由 Flask 自动托管，默认路径 /static/<filename>。在模板里务必用 url_for('static', filename='...') 生成 URL，不要硬编码路径。

static/
├── css/style.css
├── js/app.js
├── img/logo.png
└── favicon.ico

<!-- 模板中 -->
<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') }}" alt="logo">

表单处理全流程

from flask import Flask, request, redirect, url_for, flash, render_template

app = Flask(__name__)
app.secret_key = "change-me-in-production"    # flash 消息依赖 session，必须配密钥


@app.route("/register", methods=["GET", "POST"])
def register():
    if request.method == "POST":
        username = request.form.get("username", "").strip()
        email = request.form.get("email", "").strip()
        password = request.form.get("password", "")
        confirm = request.form.get("confirm", "")

        errors: list[str] = []
        # 服务端校验：永远不要信任前端
        if not (3 <= len(username) <= 20):
            errors.append("用户名长度需在 3-20 之间")
        if "@" not in email:
            errors.append("邮箱格式不正确")
        if len(password) < 8:
            errors.append("密码至少 8 位")
        if password != confirm:
            errors.append("两次密码不一致")

        if errors:
            for e in errors:
                flash(e, "error")
            # 回填用户已输入的内容，避免清空表单
            return render_template("register.html",
                                   username=username, email=email), 400

        # 真正入库（示例用内存列表代替数据库）
        flash("注册成功！", "success")
        return redirect(url_for("index"))      # PRG 模式：Post/Redirect/Get

    return render_template("register.html")

<!-- templates/register.html -->
{% extends "base.html" %}
{% block content %}
<h2>注册</h2>

<!-- 展示 flash 消息 -->
{% with messages = get_flashed_messages(with_categories=true) %}
  {% if messages %}
    <ul class="flashes">
      {% for category, msg in messages %}
        <li class="flash flash-{{ category }}">{{ msg }}</li>
      {% endfor %}
    </ul>
  {% endif %}
{% endwith %}

<form method="post" action="{{ url_for('register') }}">
  <label>用户名
    <input type="text" name="username" value="{{ username or '' }}" required>
  </label>
  <label>邮箱
    <input type="email" name="email" value="{{ email or '' }}" required>
  </label>
  <label>密码
    <!-- minlength 是前端辅助校验，真正的防线在服务端 -->
    <input type="password" name="password" minlength="8" required>
  </label>
  <label>确认密码
    <input type="password" name="confirm" required>
  </label>
  <!-- CSRF 令牌：Flask-WTF 会自动注入 -->
  <button type="submit">提交</button>
</form>
{% endblock %}

文件上传

import os
from pathlib import Path
from flask import request, jsonify

ALLOWED = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
MAX_SIZE = 5 * 1024 * 1024      # 5 MB
UPLOAD_DIR = Path("uploads")


@app.route("/upload", methods=["POST"])
def upload():
    if "file" not in request.files:
        return jsonify(error="没有文件字段"), 400

    file = request.files["file"]          # FileStorage 对象
    if file.filename == "":
        return jsonify(error="未选择文件"), 400

    # ❌ 绝不能信客户端提供的文件名！必须重新生成
    ext = Path(file.filename).suffix.lower()
    if ext not in ALLOWED:
        return jsonify(error=f"不支持的类型：{ext}"), 400

    # 检查真实类型（看魔数而不是扩展名）
    header = file.stream.read(16)
    file.stream.seek(0)
    if header[:8] == b"\x89PNG\r\n\x1a\n":
        real_ext = ".png"
    elif header[:3] == b"\xff\xd8\xff":
        real_ext = ".jpg"
    else:
        return jsonify(error="文件内容与声明不符"), 400

    UPLOAD_DIR.mkdir(exist_ok=True)
    safe_name = f"{os.urandom(8).hex()}{real_ext}"    # 随机名，防路径穿越
    path = UPLOAD_DIR / safe_name

    file.save(path)

    if path.stat().st_size > MAX_SIZE:
        path.unlink()
        return jsonify(error="文件过大"), 413

    return jsonify(ok=True, url=url_for("static", filename=f"uploads/{safe_name}")), 201

RESTful API 设计

REST（Representational State Transfer，表述性状态转移）是一套 API 设计风格，核心约束有六条：客户端-服务器、无状态、可缓存、统一接口、分层系统、代码按需可选。实际开发中真正落地的是「以资源为中心 + HTTP 动词表达动作 + 用状态码表达结果」。

设计对照表

api.py —— 完整 CRUD API

from datetime import datetime, timezone
from flask import Blueprint, jsonify, request, abort

bp = Blueprint("api", __name__)

# 内存「数据库」
DB: dict[int, dict] = {}
_next_id = 1


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def error(status: int, message: str, code: str = "error"):
    """统一的错误响应格式。"""
    response = jsonify({"error": {"code": code, "message": message}})
    response.status_code = status
    return response


# GET /api/v1/users —— 列表（分页 + 过滤 + 排序）
@bp.get("/users")
def list_users():
    page = request.args.get("page", 1, type=int)
    size = request.args.get("size", 20, type=int)
    size = max(1, min(size, 100))            # 限制单页最大条数，防资源耗尽
    page = max(1, page)

    kw = (request.args.get("kw") or "").strip().lower()
    items = list(DB.values())
    if kw:
        items = [u for u in items if kw in u["name"].lower()]

    sort = request.args.get("sort", "id")
    if sort not in {"id", "name", "created_at"}:   # 白名单，防注入
        return error(400, f"不支持的排序字段：{sort}", "bad_sort")
    items.sort(key=lambda u: u[sort], reverse=True)

    total = len(items)
    start = (page - 1) * size
    data = items[start:start + size]

    return jsonify({
        "data": data,
        "meta": {"page": page, "size": size, "total": total, "pages": (total + size - 1) // size},
    })


# GET /api/v1/users/<int:uid> —— 详情
@bp.get("/users/<int:uid>")
def get_user(uid: int):
    user = DB.get(uid)
    if user is None:
        return error(404, "用户不存在", "not_found")
    return jsonify(user)


# POST /api/v1/users —— 创建
@bp.post("/users")
def create_user():
    global _next_id
    payload = request.get_json(silent=True)
    if payload is None:
        return error(400, "请求体必须是合法 JSON", "bad_json")

    name = str(payload.get("name", "")).strip()
    if not (1 <= len(name) <= 50):
        return error(422, "name 长度需在 1-50 之间", "validation_error")

    uid = _next_id
    _next_id += 1
    user = {"id": uid, "name": name, "email": payload.get("email", ""), "created_at": now_iso()}
    DB[uid] = user

    resp = jsonify(user)
    resp.status_code = 201                       # Created
    resp.headers["Location"] = f"/api/v1/users/{uid}"   # 告诉客户端新资源地址
    return resp


# PUT /api/v1/users/<int:uid> —— 整体替换
@bp.put("/users/<int:uid>")
def replace_user(uid: int):
    if uid not in DB:
        return error(404, "用户不存在", "not_found")
    payload = request.get_json(silent=True) or {}
    name = str(payload.get("name", "")).strip()
    if not name:
        return error(422, "name 不能为空", "validation_error")
    DB[uid] = {"id": uid, "name": name, "email": payload.get("email", ""), "created_at": DB[uid]["created_at"]}
    return jsonify(DB[uid])


# PATCH /api/v1/users/<int:uid> —— 局部更新
@bp.patch("/users/<int:uid>")
def patch_user(uid: int):
    if uid not in DB:
        return error(404, "用户不存在", "not_found")
    payload = request.get_json(silent=True) or {}
    for field in ("name", "email"):              # 只允许改白名单字段
        if field in payload:
            DB[uid][field] = payload[field]
    return jsonify(DB[uid])


# DELETE /api/v1/users/<int:uid> —— 删除
@bp.delete("/users/<int:uid>")
def delete_user(uid: int):
    if DB.pop(uid, None) is None:
        return error(404, "用户不存在", "not_found")
    return "", 204                              # No Content，无响应体

统一错误响应

@bp.errorhandler(400)
def bad_request(e):
    return error(400, "请求格式错误", "bad_request")

@bp.app_errorhandler(Exception)
def on_exception(e):
    # 记录到日志，返回标准格式，绝不把堆栈暴露给客户端
    app.logger.exception("未处理异常: %s", e)
    return error(500, "服务器内部错误", "internal_error")

蓝图：大型项目的组织方式

一个文件写不下上千个路由。蓝图（Blueprint）把一组相关视图函数注册为一个可复用单元，多个蓝图可注册到同一应用，也可以嵌套挂载。

# blueprints/main.py
from flask import Blueprint, render_template

# url_prefix 会加在蓝图内所有路由前面
bp = Blueprint("main", __name__, url_prefix="", template_folder="templates")


@bp.get("/")
def index():
    return render_template("index.html")


@bp.get("/about")
def about():
    return "关于"


# 蓝图也能注册自己的蓝图（嵌套）
from .admin import bp as admin_bp
bp.register_blueprint(admin_bp, url_prefix="/admin")     # -> /admin/xxx

# blueprints/admin.py
from flask import Blueprint, request, jsonify

bp = Blueprint("admin", __name__)


@bp.get("/dashboard")
def dashboard():
    return "管理后台"


# 带 url_prefix 注册
# app.register_blueprint(admin_bp, url_prefix="/admin")
# -> /admin/dashboard

# 同一蓝图注册多次并给不同 url_prefix
# app.register_blueprint(admin_bp, url_prefix="/admin", name="admin_api")
# app.register_blueprint(admin_bp, url_prefix="/v2/admin", name="admin_v2")
# 蓝图内用 url_for 要显式写 endpoint 名：url_for("admin_api.dashboard")

# 蓝图内的扩展与上下文
bp = Blueprint("blog", __name__)

bp.before_request          # 蓝图级请求前钩子
def check_auth():
    if request.path.startswith("/api") and not is_logged_in():
        return jsonify(error="未登录"), 401

bp.after_request           # 蓝图级响应后钩子
def add_header(resp):
    resp.headers["X-Blog"] = "1"
    return resp

bp.teardown_appcontext     # 蓝图级上下文清理
def close_db(exc):
    ...

@bp.app_context_processor    # 只对蓝图模板生效
def inject_blog():
    return {"sidebar": get_sidebar()}

@bp.route("/ping")
def ping():
    return "pong"

错误处理与自定义响应

from flask import render_template, jsonify, request, make_response

# 1) abort：主动抛出标准 HTTP 异常
@app.get("/admin")
def admin():
    abort(403)                    # 等价于 raise Forbidden()
    return "永远到不了"

@app.get("/old")
def old():
    abort(410)                    # Gone，资源已永久移除
    return "到不了"


# 2) 全局错误处理器
@app.errorhandler(404)
def not_found(e):
    # API 请求返回 JSON，页面请求返回 HTML
    if request.accept_mimetypes.best == "application/json" or request.path.startswith("/api/"):
        return jsonify(error="资源不存在", code=404), 404
    return render_template("404.html"), 404

@app.errorhandler(500)
def server_error(e):
    app.logger.error("500: %s", e)
    return render_template("500.html"), 500

@app.errorhandler(Exception)
def catch_all(e):
    app.logger.exception("未捕获异常: %s", e)
    if app.config["DEBUG"]:
        raise e                  # 开发模式让 Werkzeug 调试器接管
    return jsonify(error="服务器开小差了，请稍后重试"), 500


# 3) 自定义 HTTP 状态码的响应
@app.get("/no-content")
def no_content():
    return "", 204                 # 裸空响应


# 4) make_response：精细控制状态码与响应头
@app.get("/download")
def download():
    body = "id,name\n1,alice\n"
    resp = make_response(body)
    resp.status_code = 200
    resp.headers["Content-Type"] = "text/csv; charset=utf-8"
    resp.headers["Content-Disposition"] = "attachment; filename=users.csv"
    resp.headers["Cache-Control"] = "no-store"
    return resp


# 5) 重定向与自定义响应对象
@app.get("/old-page")
def old_page():
    resp = make_response(render_template("moved.html"), 301)   # 永久重定向
    resp.headers["Location"] = "/new-page"
    return resp


# 6) 自定义异常类
class PaymentRequired(Exception):
    code = 402

@app.errorhandler(PaymentRequired)
def handle_payment(e):
    return jsonify(error="需要付费", code=e.code), e.code

@app.post("/vip")
def vip():
    raise PaymentRequired()

Web 安全基础

CSRF：跨站请求伪造

攻击者在自己的站点放一个自动提交的表单，用户登录着你的网站时访问了恶意页，浏览器会带着你的 Cookie 发起请求——于是被冒名操作。防线是令牌（token）：服务端在表单里放一个随机串，提交时校验它。

pip install flask-wtf

from flask_wtf import CSRFProtect

csrf = CSRFProtect(app)     # 之后所有 POST/PUT/PATCH/DELETE 自动校验

# 全局关闭（仅限纯 JSON API 且用 Bearer Token 认证的场景）
# CSRFProtect(app, exempt=Blueprint('api', __name__, csrf_exempt=True))

XSS：跨站脚本

把用户输入原样渲染到 HTML 中，攻击者就能注入 <script>。三道防线：① Jinja2 自动转义（默认开启，别用 |safe）；② 富文本用 bleach 清洗；③ 配置 CSP 响应头。

import re
from markupsafe import escape

# ❌ 危险：拼接 HTML
html = f"<h1>{user_input}</h1>"

# ✅ 安全：Jinja2 模板里 {{ }} 自动转义，不要用 {% autoescape false %}

# ✅ 确实需要输出 HTML 时，先清洗白名单
from bleach import clean
ALLOWED_TAGS = {"p", "b", "i", "em", "strong", "a", "ul", "li", "br", "code", "pre"}
clean_html = clean(user_input, tags=ALLOWED_TAGS, attributes={"a": ["href"]}, strip=True)

# ✅ Content Security Policy 响应头（Helmet 的 Python 版本 flask-talisman）
from flask_talisman import Talisman
Talisman(app,
         content_security_policy={
             "default-src": "'self'",
             "script-src": "'self'",       # 禁止内联脚本和第三方脚本
             "style-src": "'self' 'unsafe-inline'",
         })

其他安全配置

@app.after_request
def set_security_headers(resp):
    resp.headers.setdefault("X-Content-Type-Options", "nosniff")   # 防 MIME 嗅探
    resp.headers.setdefault("X-Frame-Options", "DENY")              # 防点击劫持
    resp.headers.setdefault("Referrer-Policy", "no-referrer")      # 防 referer 泄露
    return resp


# 生产环境强制 HTTPS
@app.before_request
def force_https():
    if request.is_secure:
        return None
    return redirect(url_for(request.endpoint, **request.view_args), code=301, _scheme="https")

安全清单

Session 与登录态

from flask import session, redirect, url_for, g, request
import secrets, hashlib

app.config.update(
    SESSION_COOKIE_HTTPONLY=True,      # JS 读不到
    SESSION_COOKIE_SECURE=True,        # 只走 HTTPS
    SESSION_COOKIE_SAMESITE="Lax",     # 防 CSRF 的一半
    PERMANENT_SESSION_LIFETIME=timedelta(days=7),
)


def hash_password(pw: str, salt: bytes | None = None) -> str:
    """PBKDF2 密码哈希：绝不要用 md5/sha1 直接哈希密码。"""
    salt = salt or secrets.token_bytes(16)
    dk = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt, 600_000)
    return f"pbkdf2_sha256$600000${salt.hex()}${dk.hex()}"


def verify_password(pw: str, stored: str) -> bool:
    try:
        algo, iters, salt_hex, hash_hex = stored.split("$")
        dk = hashlib.pbkdf2_hmac("sha256", pw.encode(), bytes.fromhex(salt_hex), int(iters))
        return secrets.compare_digest(dk.hex(), hash_hex)     # 恒定时间比较，防时序攻击
    except (ValueError, TypeError):
        return False


@app.post("/login")
def do_login():
    username = request.form.get("username", "")
    password = request.form.get("password", "")
    user = find_user(username)
    if user is None or not verify_password(password, user["password_hash"]):
        return render_template("login.html", error="用户名或密码错误"), 401

    session.clear()          # 关键：防止会话固定攻击
    session["user_id"] = user["id"]
    session.permanent = True
    return redirect(url_for("index"))


@app.before_request
def load_logged_in_user():
    """把当前用户挂到 g 上，每个请求都能用。"""
    user_id = session.get("user_id")
    g.user = get_user(user_id) if user_id else None


@app.post("/logout")
def logout():
    session.clear()
    return redirect(url_for("index"))


@app.get("/me")
def me():
    if not getattr(g, "user", None):
        return jsonify(error="未登录"), 401
    return jsonify(g.user)

工程化：配置、日志、部署

# gunicorn：生产 WSGI 服务器（Flask 自带的 dev server 只用于开发）
pip install gunicorn
gunicorn -w 4 -b 127.0.0.1:8000 --access-logfile - --error-logfile - "run:app"

# -w 4 = 4 个 worker 进程
# 生产建议配合 Nginx 做反向代理 + 静态文件托管 + TLS 终结
# gunicorn -w $(nproc) -k gthread --threads 4 -b 0.0.0.0:8000 run:app

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler("app.log", maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s [%(module)s] %(message)s"))
app.logger.addHandler(handler)
app.logger.setLevel(logging.INFO)

@app.get("/api/v1/health")
def health():
    return jsonify(status="ok", version="1.0.0")


# 请求耗时统计
@app.before_request
def before():
    g.t0 = time.perf_counter()

@app.after_request
def after(resp):
    dt = (time.perf_counter() - g.t0) * 1000
    app.logger.info("%s %s -> %s (%.1f ms)",
                    request.method, request.path, resp.status_code, dt)
    resp.headers["X-Response-Time"] = f"{dt:.1f}ms"
    return resp


# 环境变量配置（12 种方式，优先级从低到高）
# 1) 默认值 2) app.config 3) 环境变量 FLASK_ 4) .env 文件 5) 命令行 --host

# requirements.txt —— 锁定依赖
# Flask==3.0.3
# Werkzeug==3.0.3
# Jinja2==3.1.4
# Flask-SQLAlchemy==3.1.1
# flask-wtf==1.2.1
# gunicorn==22.0.0

# .env 示例（配合 python-dotenv）
# FLASK_ENV=production
# SECRET_KEY=从密钥管理服务读取，不要提交进仓库

本章小结

• HTTP 是无状态的请求-响应协议；状态码分 1xx~5xx 五类，401 是未认证、403 是无权限，GET 绝不能改数据。
• WSGI（PEP 3333）规定应用是可调用对象 app(environ, start_response)，Werkzeug 的开发服务器和 Gunicorn 都是 WSGI 服务器。
• Flask = Werkzeug + Jinja2，核心极简；推荐用应用工厂 create_app() 组织项目，便于测试和多环境配置。
• 路由用 @app.route，支持类型转换器（int/float/path/uuid/slug），可用 url_for 反向生成 URL 避免硬编码路径。
• Jinja2 模板默认自动转义变量（防 XSS）；配合 {% extends %}、{% include %}、宏（macro）复用结构。
• 表单处理要永远做服务端校验，用 PRG（Post/Redirect/Get）模式；文件上传必须校验类型、重命名、限大小。
• RESTful API：以资源为中心，动词表达动作，状态码如实反映结果，错误响应统一 JSON 格式。
• 蓝图把相关路由分组，支持 url_prefix、嵌套、蓝图级钩子，适合大型项目组织。
• 错误处理用 abort() 抛异常、errorhandler 统一渲染；API 与页面可返回不同格式。
• 安全三大件：CSRF（Flask-WTF）、XSS（Jinja2 自动转义 + CSP）、SQL 注入（参数化查询）；密码用 PBKDF2/bcrypt 哈希，永不 MD5。
• 部署用 Gunicorn 多进程 + Nginx 反向代理；SECRET_KEY 等敏感配置走环境变量。

练习题

1. 最小 WSGI 应用：手写一个符合 WSGI 规范的 app，要求：GET /hello?name=xx 返回「你好，xx」，POST 时回显请求体 JSON。分别用 werkzeug 的 run_simple 和 gunicorn 启动验证。
   参考答案要点：从 environ["QUERY_STRING"] 用 urllib.parse.parse_qs 解析参数；POST 时按 CONTENT_LENGTH 读 wsgi.input；start_response 必须先调用再返回可迭代对象。gunicorn 启动：gunicorn -b 127.0.0.1:8000 wsgi_app:app。
2. 路由与类型转换器：实现 /user/<int:uid>、/tag/<slug:tag>、/files/<path:p>、/rate/<float:r> 四个路由，分别测试匹配与不匹配的 URL，观察返回结果；然后写一个自定义 list 转换器支持 /ids/1,2,3。
   参考答案要点：slug 转换器需在 Flask 中注册：app.url_map.converters["slug"] = RegexConverter，或自定义类设 regex = r"[a-z0-9-]+"。/user/abc 不匹配 int:uid，Flask 会返回 404。
3. 模板渲染实战：用 Jinja2 渲染商品列表页。传入 5 个商品（名称、价格、是否促销、标签列表），要求：按价格倒序、只显示价格 > 100 的商品、促销商品显示红色标记、金额格式化为两位小数、无数据时显示「暂无商品」。练习使用 {% extends %}、{% block %}、{% include %}、宏和过滤器。
   参考答案要点：过滤用 {% for item in items if item.price > 100 %}；格式化用 {{ '%.2f'|format(item.price) }} 或 {{ item.price|round(2) }}；排序先在 Python 侧 items.sort(key=lambda x: -x['price']) 更直接。
4. 完整 CRUD API：用 Flask 写一个 /api/todos API，支持 GET 列表（带 ?done=true/false 过滤与 ?page=&size= 分页）、POST 创建、GET/PUT/PATCH/DELETE 详情。删除返回 204，创建返回 201 并带 Location 头，参数非法返回 422 与统一错误结构。
   参考答案要点：用 @bp.get/post/put/patch/delete 装饰器（Flask 2.0+ 语法）；分页 page = request.args.get("page", 1, type=int) 并 size = max(1, min(size, 100)) 限流；错误统一走 def error(status, msg, code) 辅助函数返回 (jsonify(...), status)。
5. 表单与校验：做一个注册表单（用户名 3-20 位、邮箱格式、密码两次一致且 ≥8 位）。要求：服务端校验失败时返回 400 并把错误信息用 flash 显示、表单数据回填、成功后重定向到首页。故意提交一个 XSS 载荷（如用户名填 <script>alert(1)</script>），验证页面上它被转义而不是执行。
   参考答案要点：用 flash(e, "error") 收集全部错误而不是遇到第一个就返回；回填 render_template("form.html", username=username, email=email), 400；XSS 验证点是 Jinja2 的 {{ }} 自动转义会输出 <script>，页面源码里不会执行。
6. 蓝图重构：把第 4 题的 API 和第 5 题的表单拆成两个蓝图：api_bp（url_prefix="/api"）、web_bp（url_prefix="/web"）。在 api_bp 上挂 before_request 做统一 API 日志，在 web_bp 上挂 context_processor 注入导航菜单。
   参考答案要点：app.register_blueprint(api_bp, url_prefix="/api")；api_bp.before_request 只对蓝图内路由生效；@api_bp.context_processor 注入的变量只在本蓝图模板里可用；可以在注册时用 name="v1" 给同一蓝图第二次注册。
7. 安全加固：给现有应用加上 ① CSRF 保护（Flask-WTF）、② 安全响应头（X-Content-Type-Options、X-Frame-Options、Referrer-Policy）、③ 登录接口（PBKDF2 哈希 + 恒定时间比较 + session.clear 防会话固定）、④ 统一 404/500 页面（页面返回 HTML、/api/ 开头的返回 JSON）。写一个 check_security.py 脚本用 requests 逐项断言。
   参考答案要点：csrf = CSRFProtect(app)；@app.after_request 加头；密码 hashlib.pbkdf2_hmac("sha256", pw, salt, 600000)，校验用 hmac.compare_digest；错误处理器里用 request.path.startswith("/api/") 判断返回 JSON。
8. 接入数据库：把第 4 题的 todos 从内存字典迁移到 SQLite（使用第 17 章的 sqlite3 参数化查询或 SQLAlchemy ORM），支持按关键词搜索标题，并用 Redis 缓存详情接口 GET /api/todos/<id>（TTL 60 秒），验证第二次请求更快。修改后用 DELETE 主动删缓存。
   参考答案要点：conn.execute("SELECT * FROM todos WHERE title LIKE ?", (f"%{kw}%",))；缓存键 f"todo:{id}"，r.setex(key, 60, json.dumps(row))；更新时 conn.commit() 后 r.delete(key)。测试时用 time.perf_counter() 对比两次耗时。
9. 生产化部署：用 Gunicorn（-w 4）启动应用，前面配 Nginx 做静态文件托管、反向代理和 gzip 压缩；用 curl -I 验证响应头里有 Server: nginx；再用 ab -n 1000 -c 50 做压测，记录 QPS，并优化一处（比如给静态文件加缓存头、把 DB 查询结果缓存起来），对比优化前后的 QPS。
   参考答案要点：Nginx 配置 location /static/ { alias ...; expires 30d; } + gzip on;；ab 若未安装可用 Python concurrent.futures + requests 手写压测脚本。注意 app.run() 自带服务器只用于开发，不能上生产。