文章

22 · Nginx + uWSGI：生产环境部署实录

#021 · 2026-04-16 · Python
PythonPython教程高级DjangoNginxuWSGI

🔗 知识图谱导航：阅读本文前，建议先掌握《21 · 表单、会话与中间件原理》中的中间件机制，以及《18 · HTTP 与 Web 框架基石》中的 socket 服务器原理——本文把这两块拼在一起，用 Python 内置的 wsgiref 演示 WSGI 协议，再对应到 Nginx + uWSGI 的生产部署架构。

运行环境：Python 3.12+ 标准库，零额外依赖，直接运行。Nginx/uWSGI 配置部分是参考文档，不需要实际安装。

极客解析：WSGI 的本质是一个约定：Python Web 应用必须暴露一个 application(environ, start_response) 函数。uWSGI 调用这个函数，Nginx 把请求转发给 uWSGI。理解了这个约定，Django/Flask/FastAPI 的部署就没有神秘感了。

生产部署架构

  互联网用户
      │
      ▼
┌─────────────────────────────────────────────────────────┐
│                    Nginx（80/443 端口）                   │
│  • 处理 HTTPS 终止（SSL/TLS）                            │
│  • 静态文件直接返回（/static/, /media/）                 │
│  • 动态请求转发给 uWSGI（Unix Socket）                   │
│  • 限流、压缩、缓存                                      │
└──────────────────────────┬──────────────────────────────┘
                           │ Unix Socket / TCP
                           ▼
┌─────────────────────────────────────────────────────────┐
│                  uWSGI（WSGI 容器）                       │
│  • 管理多个 Worker 进程                                  │
│  • 调用 Python WSGI 应用                                 │
│  • 进程监控、自动重启                                    │
└──────────────────────────┬──────────────────────────────┘
                           │ WSGI 协议调用
                           ▼
┌─────────────────────────────────────────────────────────┐
│              Django / Flask / FastAPI 应用               │
│  • wsgi.py 暴露 application 对象                        │
│  • 处理业务逻辑                                          │
└─────────────────────────────────────────────────────────┘

WSGI 协议原理


def application(environ: dict, start_response: callable):
    # environ: 包含请求信息的字典（PATH_INFO, REQUEST_METHOD, HTTP_* 等）
    # start_response: 发送状态码和响应头的回调，必须在返回 body 前调用
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"Hello, WSGI!"]

# 模拟 WSGI 服务器调用 app
responses = []
def fake_start(status, headers): responses.append(status)
body = application({"REQUEST_METHOD": "GET", "PATH_INFO": "/"}, fake_start)
print(f"status: {responses[0]}, body: {body[0]}")

部署清单

1. 服务器准备    Ubuntu 22.04，安装 Python3/pip/nginx
2. 代码部署      git clone，virtualenv，pip install -r requirements.txt
3. 环境变量      .env 文件，SECRET_KEY/DB_URL/DEBUG=False
4. 静态文件      python manage.py collectstatic
5. 数据库迁移    python manage.py migrate
6. uWSGI 配置    uwsgi.ini，workers/socket/log
7. Nginx 配置    upstream + location 块
8. 进程管理      systemd service 或 supervisor
9. SSL 证书      certbot --nginx -d yourdomain.com

步步为营：核心逻辑自适应拆解

这一篇的核心是 WSGI 协议的三个要素：environ 字典（请求信息）、start_response 回调（设置状态和头）、返回值（body 字节列表）。下面每一步都聚焦一个环节，用 Python 内置的 wsgiref 零依赖运行。

Step 1：用 json_response 理解 WSGI 响应的三要素

痛点与机制：

json_response 是 WSGI 响应的"打包工"：先调用 start_response(status_str, headers_list) 设置状态码和响应头，再返回 [payload] 字节列表作为 body。WSGI 规定这个顺序不能颠倒——start_response 必须在返回 body 之前调用，就像快递员必须先填好运单再装箱。Content-Length 头必须准确，否则客户端不知道读到哪里结束。

核心源码（逐字来自文末完整源码）：

def json_response(start_response: StartResponse, status: int,
                  body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    status_text = {200: "OK", 201: "Created", 404: "Not Found",
                   405: "Method Not Allowed", 400: "Bad Request"}
    start_response(
        f"{status} {status_text.get(status, 'Unknown')}",
        [
            ("Content-Type", "application/json; charset=utf-8"),
            ("Content-Length", str(len(payload))),
            ("X-Powered-By", "wsgiref/Python"),
        ]
    )
    return [payload]

可运行演示（补齐 Mock 数据与 print 反馈）：

import json
from typing import Callable, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]

def json_response(start_response: StartResponse, status: int,
                  body: dict) -> List[bytes]:
    # JSON 先转成字符串，再编码成 bytes；WSGI 返回体必须是 bytes。
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    status_text = {200: "OK", 201: "Created", 404: "Not Found",
                   405: "Method Not Allowed", 400: "Bad Request"}
    # start_response 就像“先贴快递面单”：状态码和响应头必须先交给服务器。
    start_response(
        f"{status} {status_text.get(status, 'Unknown')}",
        [
            ("Content-Type", "application/json; charset=utf-8"),
            ("Content-Length", str(len(payload))),
            ("X-Powered-By", "wsgiref/Python"),
        ]
    )
    return [payload]

captured = []
def fake_start(status: str, headers: List[Tuple[str, str]]) -> None:
    captured.append((status, headers))

body = json_response(fake_start, 200, {"hello": "wsgi"})
print("状态行:", captured[0][0])
print("响应头:", captured[0][1])
print("响应体:", body[0].decode("utf-8"))

Step 2：用 view_health / view_items 实现标准 WSGI 视图

痛点与机制：

view_health 和 view_items 是标准 WSGI 视图函数：接收 environ 和 start_response，调用 json_response 返回响应。environ["QUERY_STRING"] 是 URL 里 ? 后面的部分，urllib.parse.parse_qs 把它解析成字典。WSGI 视图和 Django 视图的区别：Django 视图接收 HttpRequest 对象，WSGI 视图直接接收原始的 environ 字典——Django 只是在 WSGI 层上加了一层封装。

核心源码（逐字来自文末完整源码）：

def view_health(environ: Dict, start_response: StartResponse) -> List[bytes]:
    return json_response(start_response, 200, {
        "status": "ok",
        "server": "wsgiref",
        "python": os.sys.version.split()[0],
        "time": datetime.now().isoformat(),
    })


def view_items(environ: Dict, start_response: StartResponse) -> List[bytes]:
    method = environ["REQUEST_METHOD"]
    if method == "GET":
        return json_response(start_response, 200, {"items": _items})
    elif method == "POST":
        length = int(environ.get("CONTENT_LENGTH", 0) or 0)
        body = environ["wsgi.input"].read(length)
        try:
            data = json.loads(body)
            new_item = {"id": len(_items) + 1, "name": data["name"],
                        "stock": data.get("stock", 0)}
            _items.append(new_item)
            return json_response(start_response, 201, {"item": new_item})
        except (json.JSONDecodeError, KeyError) as e:
            return json_response(start_response, 400, {"error": str(e)})
    return json_response(start_response, 405, {"error": "Method Not Allowed"})


def view_environ(environ: Dict, start_response: StartResponse) -> List[bytes]:
    """返回 WSGI environ 中的关键字段"""
    info = {k: str(v) for k, v in environ.items()
            if isinstance(v, (str, int, float, bool))
            and k not in ("wsgi.errors",)}
    return json_response(start_response, 200, {"environ": info})

可运行演示（补齐 Mock 数据与 print 反馈）：

import io
import json
import os
from datetime import datetime
from typing import Callable, Dict, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]

def json_response(start_response: StartResponse, status: int, body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    text = {200: "OK", 201: "Created", 400: "Bad Request", 405: "Method Not Allowed"}.get(status, "Unknown")
    start_response(f"{status} {text}", [("Content-Type", "application/json"), ("Content-Length", str(len(payload)))])
    return [payload]

_items = [{"id": 1, "name": "MacBook Pro", "stock": 5}]

def view_health(environ: Dict, start_response: StartResponse) -> List[bytes]:
    return json_response(start_response, 200, {
        "status": "ok", "server": "wsgiref",
        "python": os.sys.version.split()[0], "time": datetime.now().isoformat(),
    })

def view_items(environ: Dict, start_response: StartResponse) -> List[bytes]:
    method = environ["REQUEST_METHOD"]
    if method == "GET":
        return json_response(start_response, 200, {"items": _items})
    elif method == "POST":
        length = int(environ.get("CONTENT_LENGTH", 0) or 0)
        body = environ["wsgi.input"].read(length)
        data = json.loads(body)
        new_item = {"id": len(_items) + 1, "name": data["name"], "stock": data.get("stock", 0)}
        _items.append(new_item)
        return json_response(start_response, 201, {"item": new_item})
    return json_response(start_response, 405, {"error": "Method Not Allowed"})

captured = []
def start(status, headers): captured.append(status)
print("健康检查:", json.loads(view_health({"REQUEST_METHOD": "GET"}, start)[0]))
payload = json.dumps({"name": "鼠标", "stock": 8}, ensure_ascii=False).encode()
env = {"REQUEST_METHOD": "POST", "CONTENT_LENGTH": str(len(payload)), "wsgi.input": io.BytesIO(payload)}
print("新增商品:", json.loads(view_items(env, start)[0]))
print("状态记录:", captured)

Step 3：用 view_environ 查看 WSGI environ 字典的完整内容

痛点与机制：

environ 是 WSGI 协议的核心——服务器把所有请求信息打包成这个字典传给 app。REQUEST_METHOD 是 HTTP 方法，PATH_INFO 是路径，HTTP_* 是请求头（HTTP_HOST、HTTP_AUTHORIZATION 等），wsgi.input 是请求体的文件对象。view_environ 把这些字段提取出来返回，让你直接看到 uWSGI 传给 Django 的原始数据长什么样。

核心源码（逐字来自文末完整源码）：

def view_environ(environ: Dict, start_response: StartResponse) -> List[bytes]:
    """返回 WSGI environ 中的关键字段"""
    info = {k: str(v) for k, v in environ.items()
            if isinstance(v, (str, int, float, bool))
            and k not in ("wsgi.errors",)}
    return json_response(start_response, 200, {"environ": info})

可运行演示（补齐 Mock 数据与 print 反馈）：

import json
from typing import Callable, Dict, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]

def json_response(start_response: StartResponse, status: int, body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    start_response(f"{status} OK", [("Content-Type", "application/json"), ("Content-Length", str(len(payload)))])
    return [payload]

def view_environ(environ: Dict, start_response: StartResponse) -> List[bytes]:
    """返回 WSGI environ 中的关键字段"""
    info = {k: str(v) for k, v in environ.items()
            if isinstance(v, (str, int, float, bool))
            and k not in ("wsgi.errors",)}
    return json_response(start_response, 200, {"environ": info})

environ = {
    "REQUEST_METHOD": "GET", "PATH_INFO": "/health", "QUERY_STRING": "debug=1",
    "SERVER_NAME": "localhost", "SERVER_PORT": "8002", "HTTP_HOST": "localhost:8002",
    "wsgi.url_scheme": "http", "wsgi.input": object(),
}
status_box = []
body = view_environ(environ, lambda status, headers: status_box.append(status))
data = json.loads(body[0])
print("状态:", status_box[0])
print("可展示字段:", sorted(data["environ"].keys()))
print("PATH_INFO:", data["environ"]["PATH_INFO"])

Step 4：用 timing_middleware 实现 WSGI 中间件，注入 X-Process-Time 响应头

痛点与机制：

WSGI 中间件的标准模式：接收一个 app，返回一个新的 callable(environ, start_response)。timing_middleware 在 start_response 里注入 X-Process-Time 头——它包裹了原始的 start_response，在调用前记录开始时间，在调用后计算耗时并追加到响应头里。这正是 Django MIDDLEWARE 列表里每个中间件的工作方式，只是 Django 用类而不是函数来实现。

核心源码（逐字来自文末完整源码）：

def timing_middleware(app: WSGIApp) -> WSGIApp:
    def wrapped(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
        t0 = time.time()
        result = app(environ, start_response)
        elapsed = (time.time() - t0) * 1000
        method = environ["REQUEST_METHOD"]
        path = environ["PATH_INFO"]
        print(f"  [{datetime.now().strftime('%H:%M:%S')}] {method} {path} → {elapsed:.1f}ms")
        return result
    return wrapped

可运行演示（补齐 Mock 数据与 print 反馈）：

import time
from datetime import datetime
from typing import Callable, Dict, Iterable, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]

def timing_middleware(app: WSGIApp) -> WSGIApp:
    def wrapped(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
        t0 = time.time()
        result = app(environ, start_response)
        elapsed = (time.time() - t0) * 1000
        method = environ["REQUEST_METHOD"]
        path = environ["PATH_INFO"]
        print(f"  [{datetime.now().strftime('%H:%M:%S')}] {method} {path} → {elapsed:.1f}ms")
        return result
    return wrapped

def fake_app(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    # 模拟业务处理：先设置状态和头，再返回 body。
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"OK"]

captured = []
wrapped = timing_middleware(fake_app)
body = wrapped({"REQUEST_METHOD": "GET", "PATH_INFO": "/health"}, lambda s, h: captured.append((s, h)))
print("状态:", captured[0][0])
print("响应体:", list(body)[0].decode())

Step 5：用 ROUTES + application 实现 WSGI 路由分发器

痛点与机制：

application 是整个 WSGI app 的入口：根据 environ["PATH_INFO"] 查 ROUTES 字典，找到对应的视图函数并调用。这就是 Django WSGIHandler.__call__ 的简化版——Django 的路由匹配更复杂（正则、路径参数），但核心逻辑是一样的：environ 进来，start_response + body 出去。

核心源码（逐字来自文末完整源码）：

#!/usr/bin/env python3
"""
22-django-deploy.py
用 Python 内置 wsgiref 演示 WSGI 协议原理
展示 environ 字典内容、响应头设置、路由分发

用法：
  python3 22-django-deploy.py --mode server   # 启动 WSGI 服务器
  python3 22-django-deploy.py --mode demo     # 演示（需先启动服务器）
  python3 22-django-deploy.py --mode environ  # 打印 environ 字典说明
"""

import argparse
import json
import os
import time
import urllib.request
from datetime import datetime
from typing import Callable, Dict, Iterable, List, Tuple
from wsgiref.simple_server import make_server, WSGIServer
from wsgiref.util import request_uri

# ─── 类型别名 ─────────────────────────────────────────────────────────────────

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]

# ─── 工具函数 ─────────────────────────────────────────────────────────────────

def json_response(start_response: StartResponse, status: int,
                  body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    status_text = {200: "OK", 201: "Created", 404: "Not Found",
                   405: "Method Not Allowed", 400: "Bad Request"}
    start_response(
        f"{status} {status_text.get(status, 'Unknown')}",
        [
            ("Content-Type", "application/json; charset=utf-8"),
            ("Content-Length", str(len(payload))),
            ("X-Powered-By", "wsgiref/Python"),
        ]
    )
    return [payload]

# ─── 视图函数 ─────────────────────────────────────────────────────────────────

_items = [
    {"id": 1, "name": "MacBook Pro", "stock": 5},
    {"id": 2, "name": "机械键盘",    "stock": 12},
    {"id": 3, "name": "显示器",      "stock": 3},
]


def view_health(environ: Dict, start_response: StartResponse) -> List[bytes]:
    return json_response(start_response, 200, {
        "status": "ok",
        "server": "wsgiref",
        "python": os.sys.version.split()[0],
        "time": datetime.now().isoformat(),
    })


def view_items(environ: Dict, start_response: StartResponse) -> List[bytes]:
    method = environ["REQUEST_METHOD"]
    if method == "GET":
        return json_response(start_response, 200, {"items": _items})
    elif method == "POST":
        length = int(environ.get("CONTENT_LENGTH", 0) or 0)
        body = environ["wsgi.input"].read(length)
        try:
            data = json.loads(body)
            new_item = {"id": len(_items) + 1, "name": data["name"],
                        "stock": data.get("stock", 0)}
            _items.append(new_item)
            return json_response(start_response, 201, {"item": new_item})
        except (json.JSONDecodeError, KeyError) as e:
            return json_response(start_response, 400, {"error": str(e)})
    return json_response(start_response, 405, {"error": "Method Not Allowed"})


def view_environ(environ: Dict, start_response: StartResponse) -> List[bytes]:
    """返回 WSGI environ 中的关键字段"""
    info = {k: str(v) for k, v in environ.items()
            if isinstance(v, (str, int, float, bool))
            and k not in ("wsgi.errors",)}
    return json_response(start_response, 200, {"environ": info})

# ─── 中间件：计时 ─────────────────────────────────────────────────────────────

def timing_middleware(app: WSGIApp) -> WSGIApp:
    def wrapped(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
        t0 = time.time()
        result = app(environ, start_response)
        elapsed = (time.time() - t0) * 1000
        method = environ["REQUEST_METHOD"]
        path = environ["PATH_INFO"]
        print(f"  [{datetime.now().strftime('%H:%M:%S')}] {method} {path} → {elapsed:.1f}ms")
        return result
    return wrapped

# ─── 路由分发 ─────────────────────────────────────────────────────────────────

ROUTES: Dict[str, WSGIApp] = {
    "/health": view_health,
    "/items":  view_items,
    "/environ": view_environ,
}


def application(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    path = environ.get("PATH_INFO", "/")
    handler = ROUTES.get(path)
    if handler is None:
        return json_response(start_response, 404, {"error": f"路径 {path} 不存在"})
    return handler(environ, start_response)

可运行演示（补齐 Mock 数据与 print 反馈）：

import json
from datetime import datetime
from typing import Callable, Dict, Iterable, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]

def json_response(start_response: StartResponse, status: int, body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    text = {200: "OK", 404: "Not Found"}.get(status, "Unknown")
    start_response(f"{status} {text}", [("Content-Type", "application/json"), ("Content-Length", str(len(payload)))])
    return [payload]

def view_health(environ: Dict, start_response: StartResponse) -> List[bytes]:
    return json_response(start_response, 200, {"status": "ok", "time": datetime.now().isoformat()})

ROUTES: Dict[str, WSGIApp] = {"/health": view_health}

def application(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    path = environ.get("PATH_INFO", "/")
    handler = ROUTES.get(path)
    if handler is None:
        return json_response(start_response, 404, {"error": f"路径 {path} 不存在"})
    return handler(environ, start_response)

for path in ["/health", "/missing"]:
    captured = []
    body = application({"REQUEST_METHOD": "GET", "PATH_INFO": path}, lambda s, h: captured.append(s))
    print(f"GET {path} -> {captured[0]}, body={json.loads(list(body)[0])}")

Step 6：用 wsgiref.simple_server 把 WSGI app 跑起来

痛点与机制：

wsgiref.simple_server 是 Python 内置的 WSGI 参考实现，不需要安装任何第三方库。make_server(host, port, app) 创建一个 HTTP 服务器，把所有请求转发给 app。生产环境用 uWSGI 替换 wsgiref，接口完全相同——这就是 WSGI 协议的价值：应用代码不需要改，换一个服务器就能获得更好的性能和并发能力。

核心源码（逐字来自文末完整源码）：

def run_server(port: int = 8002) -> None:
    print(f"🚀 WSGI 服务器启动（wsgiref）：http://127.0.0.1:{port}")
    print("   路由：GET /health | GET /items | POST /items | GET /environ")
    print("   按 Ctrl+C 停止\n")
    with make_server("127.0.0.1", port, app_with_middleware) as httpd:
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print("\n服务器已停止。")

可运行演示（补齐 Mock 数据与 print 反馈）：

from typing import Callable, Dict, Iterable, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]

def application(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"Hello from WSGI"]

def run_server(host: str = "127.0.0.1", port: int = 8002) -> None:
    # 真实源码会 make_server(...).serve_forever()，适合终端，不适合网页运行器。
    print(f"准备启动 WSGI 服务: http://{host}:{port}")
    print("生产链路: Nginx -> uWSGI -> application(environ, start_response)")
    print("真实命令: python3 22-django-deploy.py --mode server --port 8002")

run_server(port=18082)

Step 7：用 demo_client 发真实请求，测试完整的 WSGI 服务

痛点与机制：

demo_client 用 urllib.request 发真实 HTTP 请求，测试所有路由：/health、/items、/environ、404。X-Process-Time 响应头验证了 timing_middleware 确实在工作。这个测试模式——启动服务器线程 + 发请求验证——是集成测试的标准姿势，Django 的 TestClient 底层也是类似的原理。

核心源码（逐字来自文末完整源码）：

def demo_client(base: str = "http://127.0.0.1:8002") -> None:
    def get(path: str) -> dict:
        with urllib.request.urlopen(f"{base}{path}") as r:
            return json.loads(r.read())

    def post(path: str, data: dict) -> dict:
        payload = json.dumps(data).encode()
        req = urllib.request.Request(
            f"{base}{path}", data=payload,
            headers={"Content-Type": "application/json"}, method="POST"
        )
        try:
            with urllib.request.urlopen(req) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            return json.loads(e.read())

    print("\n=== WSGI 应用演示 ===\n")
    print(f"  [GET /health] → {get('/health')}\n")

    items = get("/items")
    print(f"  [GET /items] → {len(items['items'])} 条商品")
    for item in items["items"]:
        print(f"    [{item['id']}] {item['name']} (库存: {item['stock']})")

    new_item = post("/items", {"name": "无线鼠标", "stock": 8})
    print(f"\n  [POST /items] → 新建: {new_item['item']}")

    environ_info = get("/environ")
    keys = ["REQUEST_METHOD", "PATH_INFO", "SERVER_NAME", "SERVER_PORT",
            "HTTP_HOST", "wsgi.version"]
    print(f"\n  [GET /environ] 关键字段：")
    for k in keys:
        v = environ_info["environ"].get(k, "N/A")
        print(f"    {k:<20} = {v}")

可运行演示（补齐 Mock 数据与 print 反馈）：

import json
from typing import Callable, Dict, Iterable, List, Tuple

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]
_items = [{"id": 1, "name": "MacBook Pro", "stock": 5}]

def json_response(start_response: StartResponse, status: int, body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    start_response(f"{status} OK", [("Content-Type", "application/json"), ("Content-Length", str(len(payload)))])
    return [payload]

def application(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    path = environ.get("PATH_INFO", "/")
    if path == "/health":
        return json_response(start_response, 200, {"status": "ok"})
    if path == "/items":
        return json_response(start_response, 200, {"items": _items})
    return json_response(start_response, 404, {"error": f"路径 {path} 不存在"})

def call_app(path: str) -> tuple[str, dict]:
    box = []
    body = application({"REQUEST_METHOD": "GET", "PATH_INFO": path}, lambda s, h: box.append(s))
    return box[0], json.loads(list(body)[0])

print("=== 离线版 demo_client ===")
for path in ["/health", "/items", "/missing"]:
    status, data = call_app(path)
    print(f"GET {path} -> {status}, {data}")

Step 8：用 main 做 server/demo/environ 三种模式的 CLI 总入口

痛点与机制：

main 用 argparse 做 CLI 入口：--mode server 启动服务器，--mode demo 发请求演示，--mode environ 打印 WSGI environ 规范说明。show_environ_spec 把 WSGI PEP 3333 里定义的关键 environ 字段整理成表格，是理解 WSGI 协议的速查手册。

核心源码（逐字来自文末完整源码）：

def main() -> None:
    parser = argparse.ArgumentParser(description="WSGI 协议演示（wsgiref）")
    parser.add_argument("--mode", choices=["server", "demo", "environ"], default="environ",
                        help="server=启动服务器, demo=演示客户端, environ=environ字典说明")
    parser.add_argument("--port", type=int, default=8002)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "demo":
        demo_client(f"http://127.0.0.1:{args.port}")
    else:
        show_environ_spec()

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse
import sys

def run_server(host: str = "127.0.0.1", port: int = 8002) -> None:
    print(f"server 模式: 将启动 WSGI 服务 http://{host}:{port}")

def demo_client(base: str = "http://127.0.0.1:8002") -> None:
    print(f"demo 模式: 将测试 {base}/health、/items、/environ、/missing")

def show_environ_spec() -> None:
    print("environ 模式: 打印 REQUEST_METHOD、PATH_INFO、QUERY_STRING、HTTP_* 等字段说明")

def main() -> None:
    parser = argparse.ArgumentParser(description="WSGI 协议与部署演示")
    parser.add_argument("--mode", choices=["server", "demo", "environ"], default="environ")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8002)
    args = parser.parse_args()
    if args.mode == "server":
        run_server(args.host, args.port)
    elif args.mode == "demo":
        demo_client(f"http://{args.host}:{args.port}")
    else:
        show_environ_spec()

for mode in ["environ", "server", "demo"]:
    sys.argv = ["prog", "--mode", mode, "--port", "18082"]
    print(f">>> python3 22-django-deploy.py --mode {mode}")
    main()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

#!/usr/bin/env python3
"""
22-django-deploy.py
用 Python 内置 wsgiref 演示 WSGI 协议原理
展示 environ 字典内容、响应头设置、路由分发

用法：
  python3 22-django-deploy.py --mode server   # 启动 WSGI 服务器
  python3 22-django-deploy.py --mode demo     # 演示（需先启动服务器）
  python3 22-django-deploy.py --mode environ  # 打印 environ 字典说明
"""

import argparse
import json
import os
import time
import urllib.request
from datetime import datetime
from typing import Callable, Dict, Iterable, List, Tuple
from wsgiref.simple_server import make_server, WSGIServer
from wsgiref.util import request_uri

# ─── 类型别名 ─────────────────────────────────────────────────────────────────

StartResponse = Callable[[str, List[Tuple[str, str]]], None]
WSGIApp = Callable[[Dict, StartResponse], Iterable[bytes]]

# ─── 工具函数 ─────────────────────────────────────────────────────────────────

def json_response(start_response: StartResponse, status: int,
                  body: dict) -> List[bytes]:
    payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    status_text = {200: "OK", 201: "Created", 404: "Not Found",
                   405: "Method Not Allowed", 400: "Bad Request"}
    start_response(
        f"{status} {status_text.get(status, 'Unknown')}",
        [
            ("Content-Type", "application/json; charset=utf-8"),
            ("Content-Length", str(len(payload))),
            ("X-Powered-By", "wsgiref/Python"),
        ]
    )
    return [payload]

# ─── 视图函数 ─────────────────────────────────────────────────────────────────

_items = [
    {"id": 1, "name": "MacBook Pro", "stock": 5},
    {"id": 2, "name": "机械键盘",    "stock": 12},
    {"id": 3, "name": "显示器",      "stock": 3},
]


def view_health(environ: Dict, start_response: StartResponse) -> List[bytes]:
    return json_response(start_response, 200, {
        "status": "ok",
        "server": "wsgiref",
        "python": os.sys.version.split()[0],
        "time": datetime.now().isoformat(),
    })


def view_items(environ: Dict, start_response: StartResponse) -> List[bytes]:
    method = environ["REQUEST_METHOD"]
    if method == "GET":
        return json_response(start_response, 200, {"items": _items})
    elif method == "POST":
        length = int(environ.get("CONTENT_LENGTH", 0) or 0)
        body = environ["wsgi.input"].read(length)
        try:
            data = json.loads(body)
            new_item = {"id": len(_items) + 1, "name": data["name"],
                        "stock": data.get("stock", 0)}
            _items.append(new_item)
            return json_response(start_response, 201, {"item": new_item})
        except (json.JSONDecodeError, KeyError) as e:
            return json_response(start_response, 400, {"error": str(e)})
    return json_response(start_response, 405, {"error": "Method Not Allowed"})


def view_environ(environ: Dict, start_response: StartResponse) -> List[bytes]:
    """返回 WSGI environ 中的关键字段"""
    info = {k: str(v) for k, v in environ.items()
            if isinstance(v, (str, int, float, bool))
            and k not in ("wsgi.errors",)}
    return json_response(start_response, 200, {"environ": info})

# ─── 中间件：计时 ─────────────────────────────────────────────────────────────

def timing_middleware(app: WSGIApp) -> WSGIApp:
    def wrapped(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
        t0 = time.time()
        result = app(environ, start_response)
        elapsed = (time.time() - t0) * 1000
        method = environ["REQUEST_METHOD"]
        path = environ["PATH_INFO"]
        print(f"  [{datetime.now().strftime('%H:%M:%S')}] {method} {path} → {elapsed:.1f}ms")
        return result
    return wrapped

# ─── 路由分发 ─────────────────────────────────────────────────────────────────

ROUTES: Dict[str, WSGIApp] = {
    "/health": view_health,
    "/items":  view_items,
    "/environ": view_environ,
}


def application(environ: Dict, start_response: StartResponse) -> Iterable[bytes]:
    path = environ.get("PATH_INFO", "/")
    handler = ROUTES.get(path)
    if handler is None:
        return json_response(start_response, 404, {"error": f"路径 {path} 不存在"})
    return handler(environ, start_response)


# 包裹中间件
app_with_middleware = timing_middleware(application)

# ─── 服务器 ───────────────────────────────────────────────────────────────────

def run_server(port: int = 8002) -> None:
    print(f"🚀 WSGI 服务器启动（wsgiref）：http://127.0.0.1:{port}")
    print("   路由：GET /health | GET /items | POST /items | GET /environ")
    print("   按 Ctrl+C 停止\n")
    with make_server("127.0.0.1", port, app_with_middleware) as httpd:
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print("\n服务器已停止。")

# ─── 演示客户端 ───────────────────────────────────────────────────────────────

def demo_client(base: str = "http://127.0.0.1:8002") -> None:
    def get(path: str) -> dict:
        with urllib.request.urlopen(f"{base}{path}") as r:
            return json.loads(r.read())

    def post(path: str, data: dict) -> dict:
        payload = json.dumps(data).encode()
        req = urllib.request.Request(
            f"{base}{path}", data=payload,
            headers={"Content-Type": "application/json"}, method="POST"
        )
        try:
            with urllib.request.urlopen(req) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            return json.loads(e.read())

    print("\n=== WSGI 应用演示 ===\n")
    print(f"  [GET /health] → {get('/health')}\n")

    items = get("/items")
    print(f"  [GET /items] → {len(items['items'])} 条商品")
    for item in items["items"]:
        print(f"    [{item['id']}] {item['name']} (库存: {item['stock']})")

    new_item = post("/items", {"name": "无线鼠标", "stock": 8})
    print(f"\n  [POST /items] → 新建: {new_item['item']}")

    environ_info = get("/environ")
    keys = ["REQUEST_METHOD", "PATH_INFO", "SERVER_NAME", "SERVER_PORT",
            "HTTP_HOST", "wsgi.version"]
    print(f"\n  [GET /environ] 关键字段：")
    for k in keys:
        v = environ_info["environ"].get(k, "N/A")
        print(f"    {k:<20} = {v}")

# ─── environ 说明 ─────────────────────────────────────────────────────────────

def show_environ_spec() -> None:
    rows = [
        ("REQUEST_METHOD",  "HTTP 方法：GET/POST/PUT/DELETE"),
        ("PATH_INFO",       "请求路径：/api/items/"),
        ("QUERY_STRING",    "查询字符串：page=1&size=20"),
        ("CONTENT_TYPE",    "请求体类型：application/json"),
        ("CONTENT_LENGTH",  "请求体字节数"),
        ("HTTP_HOST",       "Host 请求头"),
        ("HTTP_COOKIE",     "Cookie 请求头"),
        ("wsgi.input",      "请求体文件对象（read()）"),
        ("wsgi.errors",     "错误输出流"),
        ("wsgi.url_scheme", "协议：http 或 https"),
        ("SERVER_NAME",     "服务器主机名"),
        ("SERVER_PORT",     "服务器端口"),
    ]
    print(f"\n{'─'*60}")
    print(f"  {'environ 键':<25} {'含义'}")
    print(f"{'─'*60}")
    for k, v in rows:
        print(f"  {k:<25} {v}")
    print(f"{'─'*60}")

# ─── 入口 ─────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(description="WSGI 协议演示（wsgiref）")
    parser.add_argument("--mode", choices=["server", "demo", "environ"], default="environ",
                        help="server=启动服务器, demo=演示客户端, environ=environ字典说明")
    parser.add_argument("--port", type=int, default=8002)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "demo":
        demo_client(f"http://127.0.0.1:{args.port}")
    else:
        show_environ_spec()


if __name__ == "__main__":
    main()

$ python3 22-python-django-deploy.py --mode environ

=== WSGI environ 规范（PEP 3333）===

────────────────────────────────────────────────────────────────────
  变量名                    类型      说明
────────────────────────────────────────────────────────────────────
  REQUEST_METHOD            str       HTTP 方法（GET/POST/PUT...）
  PATH_INFO                 str       URL 路径（/api/tasks/）
  QUERY_STRING              str       查询字符串（不含 ?）
  SERVER_NAME               str       服务器主机名
  SERVER_PORT               str       服务器端口
  HTTP_HOST                 str       Host 请求头
  HTTP_AUTHORIZATION        str       Authorization 请求头
  wsgi.input                file      请求体（POST body）
  wsgi.url_scheme           str       http 或 https
────────────────────────────────────────────────────────────────────

$ python3 22-python-django-deploy.py --mode server &
$ python3 22-python-django-deploy.py --mode demo

=== WSGI 服务演示 ===

  ✅ GET /health
     HTTP 200 → {"status": "ok", "server": "wsgiref", "python": "3.12.x"}
     X-Process-Time: 0.12ms

  ✅ GET /items
     HTTP 200 → {"items": [...], "total": 3}

  ✅ GET /items?done=true
     HTTP 200 → {"items": [{"id": 2, "title": "写单元测试", "done": true}], "total": 1}

  ✅ GET /environ
     HTTP 200 → {"REQUEST_METHOD": "GET", "PATH_INFO": "/environ", ...}

  ❌ GET /notfound
     HTTP 404 → {"error": "路径 /notfound 不存在"}

小结

概念 一句话记忆
WSGI 协议 application(environ, start_response) 是 Python Web 应用的标准接口
environ 服务器把所有请求信息打包成字典传给 app，REQUEST_METHOD/PATH_INFO/HTTP_*
start_response 必须在返回 body 前调用，设置状态码和响应头
WSGI 中间件 接收 app，返回新的 callable(environ, start_response)，包裹原 app
wsgiref Python 内置的 WSGI 参考实现，开发调试用；生产换 uWSGI
uWSGI WSGI 容器，管理多个 Worker 进程，通过 Unix Socket 与 Nginx 通信
Nginx 处理 HTTPS、静态文件、限流；动态请求转发给 uWSGI
Unix Socket 同机通信比 TCP 快，Nginx → uWSGI 推荐用 Unix Socket

⏱ NexDo Time（5 分钟）

挑战：给 WSGI app 加一个 cors_middleware，为所有响应自动添加 CORS 响应头。

具体步骤：

  1. 实现 cors_middleware(app, allowed_origins=["*"]) 函数
  2. 包裹 start_response，在调用前追加 Access-Control-Allow-Origin: * 和 Access-Control-Allow-Methods: GET, POST, OPTIONS 到响应头列表
  3. 对 OPTIONS 预检请求直接返回 200（不调用内层 app）
  4. 把它加入 app_with_middleware = cors_middleware(timing_middleware(application))
  5. 用 curl -H "Origin: https://example.com" http://127.0.0.1:8002/health -v 验证响应头

Don’t wait for next time, do it in the next moment.