正心全栈编程

用中间件处理公共逻辑 ​

每个接口都可能需要记录请求、统计耗时或添加统一响应头。这些工作围绕请求执行，却不属于某一篇文章的业务逻辑。本专题把它们放到中间件中，并观察多个中间件的执行顺序。

中间件包围一次请求与响应。call_next(request) 将请求继续交给后续链路，返回后还能修改响应。示例代码见下文。

中间件的执行位置 ​

先观察中间件与处理函数的执行位置，再添加实际工作。异常处理器负责错误响应，两者职责不同。

先写一个最小中间件，只用来观察它的位置。

示例文件：01_middleware_order.py。运行：uvicorn 01_middleware_order:app --reload。

py
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import PlainTextResponse
from starlette.routing import Route


class SimpleLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print("中间件：进入 endpoint 之前")
        response = await call_next(request)
        print("中间件：响应返回客户端之前")
        return response


async def homepage(request):
    print("endpoint：正在处理请求")
    return PlainTextResponse("查看控制台输出顺序")


routes = [
    Route("/", homepage),
]


middleware = [
    Middleware(SimpleLogMiddleware),
]


app = Starlette(
    routes=routes,
    middleware=middleware,
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("01_middleware_order:app", host="127.0.0.1", port=8000, reload=True)

访问 / 时，大致输出顺序会是：

text
中间件：进入 endpoint 之前
endpoint：正在处理请求
中间件：响应返回客户端之前

这正好说明了中间件的位置。

中间件洋葱圈模型与双向拦截架构

注册中间件 ​

中间件通过 middleware 列表注册：

python
middleware = [
    Middleware(SimpleLogMiddleware),
]

然后传给：

python
Starlette(routes=routes, middleware=middleware)

记录请求与耗时 ​

日志和耗时统计都关心整个请求流程。多个中间件嵌套执行，阅读输出时要区分进入顺序和返回顺序。

接下来写两个更实用的中间件：

  1. 请求日志
  2. 耗时统计

示例文件：02_logging_and_timing.py。运行：uvicorn 02_logging_and_timing:app --reload。

py
import time

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


class RequestLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print(f"请求开始：{request.method} {request.url.path}")
        response = await call_next(request)
        print(f"请求结束：{response.status_code}")
        return response


class ProcessTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        process_time = time.perf_counter() - start_time
        response.headers["X-Process-Time"] = f"{process_time:.6f}"
        return response


async def homepage(request):
    return PlainTextResponse("请求日志和耗时统计")


async def article_list(request):
    return JSONResponse(
        {
            "items": [
                {"id": 1, "title": "认识 Starlette 应用对象"},
                {"id": 2, "title": "学习中间件"},
            ],
        }
    )


routes = [
    Route("/", homepage),
    Route("/articles", article_list),
]


middleware = [
    Middleware(RequestLogMiddleware),
    Middleware(ProcessTimeMiddleware),
]


app = Starlette(
    routes=routes,
    middleware=middleware,
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("02_logging_and_timing:app", host="127.0.0.1", port=8000, reload=True)

请求日志的执行过程 ​

它负责打印：

text
请求开始：方法 + 路径
请求结束：响应状态码

耗时统计的执行过程 ​

它负责：

text
请求进入时记录起始时间
请求结束时计算耗时
把耗时写进响应头

中间件执行顺序 ​

多个中间件叠在一起时，可以理解为一层套一层。请求进来时按注册顺序进入，响应返回时按相反方向退出。

多中间件级联执行与全链路耗时时序图

添加统一响应头 ​

统一响应头适合在响应经过中间件时设置，避免每个终点重复编写。

再看一个很常见的场景：给所有响应统一添加一个响应头。

示例文件：03_response_headers.py。运行：uvicorn 03_response_headers:app --reload。

py
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


class HeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Course-Name"] = "Starlette"
        return response


async def homepage(request):
    return PlainTextResponse("统一响应头演示")


async def article_list(request):
    return JSONResponse(
        {
            "message": "文章列表",
            "items": [
                {"id": 1, "title": "认识 Starlette 应用对象"},
                {"id": 2, "title": "学习中间件"},
            ],
        }
    )


routes = [
    Route("/", homepage),
    Route("/articles", article_list),
]


middleware = [
    Middleware(HeaderMiddleware),
]


app = Starlette(
    routes=routes,
    middleware=middleware,
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("03_response_headers:app", host="127.0.0.1", port=8000, reload=True)

无论访问哪个接口，中间件都会在响应返回前统一加上：

text
X-Course-Name: Starlette

统一响应头的放置位置 ​

因为这类逻辑通常是“所有响应都要做”的。如果写到每个 endpoint 里，就会重复很多次。

配置 CORS 与 GZip ​

示例文件：04_cors_and_gzip.py。运行：uvicorn 04_cors_and_gzip:app --reload。

CORSMiddleware 处理跨域访问规则，GZipMiddleware 压缩符合条件的响应。配置应与实际演示请求对应。

除了自定义中间件，Starlette 也提供了一些内置中间件。

CORS 与 GZip 的作用 ​

这里可区分为：

text
CORS：处理浏览器跨域访问
GZip：压缩较大的响应内容

跨域与压缩示例 ​

py
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


async def homepage(request):
    return PlainTextResponse("CORS 和 GZip 中间件演示")


async def article_list(request):
    items = [{"id": i, "title": f"文章 {i}"} for i in range(1, 101)]
    return JSONResponse({"items": items})


routes = [
    Route("/", homepage),
    Route("/articles", article_list),
]


middleware = [
    Middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["*"],
        allow_headers=["*"],
    ),
    Middleware(GZipMiddleware, minimum_size=100),
]


app = Starlette(
    routes=routes,
    middleware=middleware,
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("04_cors_and_gzip:app", host="127.0.0.1", port=8000, reload=True)

CORSMiddleware 配置 ​

教学阶段先只记住几个核心参数：

text
allow_origins  允许哪些来源访问
allow_methods  允许哪些请求方法
allow_headers  允许哪些请求头

GZipMiddleware 配置 ​

这里最关键的是：

python
minimum_size=100

表示响应内容达到一定大小后，才会压缩。

其他内置中间件 ​

本章不展开更多中间件，只先知道：

text
有些通用能力可以自己写
有些框架已经提供现成中间件

内置中间件执行模型：CORS 预检与 GZip 压缩机制

中间件的职责边界 ​

适合处理的逻辑 ​

一般来说，这些更适合放中间件：

text
请求日志
耗时统计
统一响应头
跨域
压缩
认证前置检查（简单场景）

共同特点是：

text
很多接口都会用到
不依赖某个具体业务资源

不适合处理的逻辑 ​

这些通常不适合放中间件：

text
创建文章
修改文章
删除文章
查询某篇文章
复杂业务校验

因为这些逻辑属于具体业务，应该留在 endpoint 或服务层。

中间件与异常处理器 ​

中间件和异常处理器职责不同。

text
通用请求/响应处理 → 中间件
统一错误响应      → 异常处理器

不要混着用，也不要指望中间件取代异常处理器。

中间件与业务层、异常处理器的职责分工矩阵

文章服务中间件案例 ​

完整案例运行 05_article_middleware.py 后，可用 middleware_demo.py 发送请求，查看状态码、响应体及关键响应头。

完整案例把日志、耗时统计和响应头加到文章服务中。

目标包括：

text
记录请求日志
统计接口耗时
为所有响应添加统一响应头
保留文章列表、创建文章、文章详情等基本接口

示例文件：05_article_middleware.py。运行：uvicorn 05_article_middleware:app --reload。

py
import time

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习 Request 请求对象"},
    {"id": 3, "title": "学习 Response 响应对象"},
]


class RequestLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print(f"请求开始：{request.method} {request.url.path}")
        response = await call_next(request)
        print(f"请求结束：{response.status_code}")
        return response


class ProcessTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        process_time = time.perf_counter() - start_time
        response.headers["X-Process-Time"] = f"{process_time:.6f}"
        return response


class HeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Course-Name"] = "Starlette"
        return response


async def homepage(request):
    return PlainTextResponse("欢迎学习中间件")


async def article_list(request):
    return JSONResponse({"items": ARTICLES})


async def article_create(request):
    request_data = await request.json()
    title = request_data.get("title", "未命名文章")
    new_article = {"id": len(ARTICLES) + 1, "title": title}
    ARTICLES.append(new_article)
    return JSONResponse({"message": "文章创建成功", "article": new_article}, status_code=201)


async def article_detail(request):
    article_id = request.path_params["article_id"]
    for article in ARTICLES:
        if article["id"] == article_id:
            return JSONResponse(article)
    return JSONResponse({"message": "文章不存在"}, status_code=404)


routes = [
    Route("/", homepage),
    Route("/articles", article_list, methods=["GET"]),
    Route("/articles", article_create, methods=["POST"]),
    Route("/articles/{article_id:int}", article_detail, methods=["GET"]),
]


middleware = [
    Middleware(RequestLogMiddleware),
    Middleware(ProcessTimeMiddleware),
    Middleware(HeaderMiddleware),
]


app = Starlette(
    routes=routes,
    middleware=middleware,
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("05_article_middleware:app", host="127.0.0.1", port=8000, reload=True)

1、访问首页

访问：

text
http://127.0.0.1:8000/

2、访问文章列表

访问：

text
http://127.0.0.1:8000/articles

3、创建文章

向下面地址发送 POST：

text
http://127.0.0.1:8000/articles

4、访问不存在文章

访问：

text
http://127.0.0.1:8000/articles/999

5、请求处理流程

可以把完整流程理解成：

text
客户端请求
   ↓
RequestLogMiddleware 打印日志
   ↓
ProcessTimeMiddleware 开始计时
   ↓
进入 endpoint
   ↓
endpoint 返回响应
   ↓
ProcessTimeMiddleware 写入耗时响应头
   ↓
HeaderMiddleware 写入统一响应头
   ↓
客户端收到最终响应

案例源码 ​

py
from pprint import pprint

import requests


BASE_URL = "http://127.0.0.1:8000"


def test_homepage_headers():
    response = requests.get(f"{BASE_URL}/")
    print(f"\n=== GET / ===")
    print(f"status_code: {response.status_code}")
    print(f"body: {response.text}")
    print(f"X-Process-Time: {response.headers.get('X-Process-Time')}")
    print(f"X-Course-Name: {response.headers.get('X-Course-Name')}")


def test_article_list_headers():
    response = requests.get(f"{BASE_URL}/articles")
    print(f"\n=== GET /articles 文章列表 ===")
    print(f"status_code: {response.status_code}")
    print(f"X-Process-Time: {response.headers.get('X-Process-Time')}")
    print(f"X-Course-Name: {response.headers.get('X-Course-Name')}")
    pprint(response.json())


def test_create_article_with_headers():
    response = requests.post(
        f"{BASE_URL}/articles",
        json={"title": "中间件专题的新文章"},
    )
    print(f"\n=== POST /articles 创建文章 ===")
    print(f"status_code: {response.status_code}")
    print(f"X-Process-Time: {response.headers.get('X-Process-Time')}")
    print(f"X-Course-Name: {response.headers.get('X-Course-Name')}")
    pprint(response.json())


def test_create_second_article():
    response = requests.post(
        f"{BASE_URL}/articles",
        json={"title": "中间件专题的第二篇文章"},
    )
    print(f"\n=== POST /articles 再次创建文章 ===")
    print(f"status_code: {response.status_code}")
    print(f"X-Process-Time: {response.headers.get('X-Process-Time')}")
    print(f"X-Course-Name: {response.headers.get('X-Course-Name')}")
    pprint(response.json())


def test_article_detail_headers():
    response = requests.get(f"{BASE_URL}/articles/1")
    print(f"\n=== GET /articles/1 查看文章详情 ===")
    print(f"status_code: {response.status_code}")
    print(f"X-Process-Time: {response.headers.get('X-Process-Time')}")
    print(f"X-Course-Name: {response.headers.get('X-Course-Name')}")
    pprint(response.json())


if __name__ == "__main__":
    test_homepage_headers()
    test_article_list_headers()
    test_create_article_with_headers()
    test_create_second_article()
    test_article_detail_headers()