FastAPI教程-middleware中间件

作者：马育民 • • 阅读：10028

介绍

FastAPI中间件本质是 ASGI中间件，在请求到达路由函数之前、响应返回客户端之前做拦截处理，可以处理请求、修改请求、修改响应、异常捕获、日志、计时、跨域等通用逻辑。

FastAPI基于Starlette，中间件完全复用Starlette中间件机制

有两种写法：

  • 类中间件（传统ASGI）
  • @app.middleware("http")装饰器中间件（最常用）

理解

对于java程序员，可以理解为过滤器、拦截器

应用场景

  • 全局请求日志打印：打印请求方法、url、客户端 IP、请求头、状态码、响应耗时；统一访问日志。

  • 请求限流、IP 黑名单：拦截黑名单 IP；简单全局计数器限流，超限直接返回 429。

  • GZIP、压缩、代理转发相关：简单响应内容修改；复杂压缩优先使用内置 GZipMiddleware。

  • CORS 跨域：官方内置 CORSMiddleware，处理跨域预检 OPTIONS 请求。

  • 统一修改请求header、响应 header、状态码。

不适合

  • 鉴权：一般只有部分接口需要鉴权

执行流程

  1. 客户端发起HTTP请求 → 中间件接收请求（传入request 对象）
  2. 中间件预处理请求（如：修改header、日志、计时开始）
  3. 将请求向下传递，交给路由处理
  4. 路由执行视图函数，返回 response 响应对象
  5. 回到中间件，对 response 做后置处理（添加响应头、计算耗时）
  6. 返回 response 给客户端

实现方式1-装饰器（推荐）

普通函数上面加上装饰器：@app.middleware("http")，http 表示拦截 HTTP 请求

关键点：call_next(request) 是把请求交给后续链路，必须调用，否则请求不会走到接口；它返回的是response。

例子

from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def my_middleware1(request, call_next):
    print(f"中间件1-收到请求: {request.method}请求方法，URL：{request.url}")
    # 把请求交给下一层，执行接口，拿到响应，必须写await
    response = await call_next(request)
    print(f"中间件1-请求结束")
    return response

@app.get('/student/{name}')
async def get_student(name: str):
    return {
            "code": 200,
                "data": {
                'name': name,
                'age': 21,
            }
    }

测试

浏览器发请求：http://127.0.0.1:8000/student/李雷 ，控制台打印如下：

中间件1-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
执行路由处理函数
中间件1-请求结束

实现方式2-类实现

定义类，继承 BaseHTTPMiddleware 父类，只处理 http 请求

好处：可以封装为独立类，多个项目复用。

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

app = FastAPI()

class MyMiddleware1(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        print(f"中间件1-收到请求: {request.method}请求方法，URL：{request.url}")
        resp = await call_next(request)
        print(f"中间件1-请求结束")
        return resp

# 注册中间件
app.add_middleware(MyMiddleware1)

@app.get('/student/{name}')
async def get_student(name: str):
    print("执行路由处理函数")
    return {
            "code": 200,
                "data": {
                'name': name,
                'age': 21,
            }
    }

测试

浏览器发请求：http://127.0.0.1:8000/student/李雷 ，控制台打印如下：

中间件1-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
执行路由处理函数
中间件1-请求结束

多个中间件执行顺序-装饰器实现

装饰器 @app.middleware("http") 执行顺序：从下到上

例子

from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def my_middleware1(request, call_next):
    print(f"中间件1-收到请求: {request.method}请求方法，URL：{request.url}")
    # 把请求交给下一层，执行接口，拿到响应，必须写await
    response = await call_next(request)
    print(f"中间件1-请求结束")
    return response


@app.middleware("http")
async def my_middleware2(request, call_next):
    print(f"中间件2-收到请求: {request.method}请求方法，URL：{request.url}")
    # 把请求交给下一层，执行接口，拿到响应，必须写await
    response = await call_next(request)
    print(f"中间件2-请求结束")
    return response


@app.get('/student/{name}')
async def get_student(name: str):
    print("执行路由处理函数")
    return {
            "code": 200,
                "data": {
                'name': name,
                'age': 21,
            }
    }

测试

浏览器发请求：http://127.0.0.1:8000/student/李雷 ，控制台打印如下：

中间件2-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
中间件1-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
执行路由处理函数
中间件1-请求结束
中间件2-请求结束

从下到上：先执行下面的中间件，再执行上面的中间件

多个中间件执行顺序-类实现

add_middleware() 执行顺序：从下到上，即：后添加的先执行

例子

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

app = FastAPI()

class MyMiddleware1(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        print(f"中间件1-收到请求: {request.method}请求方法，URL：{request.url}")
        resp = await call_next(request)
        print(f"中间件1-请求结束")
        return resp


class MyMiddleware2(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        print(f"中间件2-收到请求: {request.method}请求方法，URL：{request.url}")
        resp = await call_next(request)
        print(f"中间件2-请求结束")
        return resp

# 注册中间件
app.add_middleware(MyMiddleware1)
app.add_middleware(MyMiddleware2)

@app.get('/student/{name}')
async def get_student(name: str):
    print("执行路由处理函数")
    return {
            "code": 200,
                "data": {
                'name': name,
                'age': 21,
            }
    }

测试

浏览器发请求：http://127.0.0.1:8000/student/李雷 ，控制台打印如下：

中间件2-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
中间件1-收到请求: GET请求方法，URL：http://127.0.0.1:8000/student/李雷
执行路由处理函数
中间件1-请求结束
中间件2-请求结束

从下到上，即：后添加的先执行

常见内置中间件

1. CORS跨域中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许的源，生产环境不要*
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

2. GZip压缩中间件

from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)

大于1000字节响应自动gzip压缩。

直接返回响应（拦截请求，不走路由）

不调用 call_next，直接返回 Response，请求不会进入路由函数，适合鉴权拦截

注意：一般不用中间件鉴权，而是使用依赖

from fastapi.responses import JSONResponse

@app.middleware("http")
async def auth_mw(request, call_next):
    token = request.headers.get("token")
    if not token:
        # 直接返回，不走接口
        return JSONResponse(status_code=401, content={"code":401,"msg":"未登录"})
    return await call_next(request)

原文出处：/show_1GW3s6Wj91ol.html

吉ICP备15003043号