1.介绍
1.1 官网介绍
"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.
它接收你的应用程序的每一个
请求.
然后它可以对这个
请求做一些事情或者执行任何需要的代码.
然后它将
请求传递给应用程序的其他部分 (通过某种
路径操作).
然后它获取应用程序生产的
响应 (通过某种
路径操作).
它可以对该
响应做些什么或者执行任何需要的代码.
然后它返回这个
响应.
1.2 中间件工作示意图
1.3 官方使用示例
import timefrom fastapi import FastAPI, Requestapp = FastAPI()@app.middleware("http")async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response
2. 快速使用
从官方示例可以看出，中间件函数要和FastAPI实例在一个文件才能通过注解的方式，这种虽然使用起来比较简单，但是不太合适扩展和项目结构管理，下面是通过函数add_middleware来注册中间件。
2.1 创建中间件
在包app/middleware下，并新增文件usetime_middleware.py，文件内容如下:
import timefrom fastapi import Requestfrom starlette.middleware.base import BaseHTTPMiddlewarefrom starlette.responses import Responseclass UseTimeMiddleware(BaseHTTPMiddleware): """ 计算耗时中间件""" def __init__(self, app): super().__init__(app) async def dispatch(self, request: Request, call_next) -> Response: """ 请求耗时 """ start_time = time.time() # 调用下一个中间件或路由处理函数 result = await call_next(request) process_time = time.time() - start_time result.headers["X-Process-Time"] = str(process_time) return result
@注意:我们定义的中间件类UseTimeMiddleware要继承基础类BaseHTTPMiddleware
2.2 封装注册函数
在包app/middleware/__init__.py引用,并封装统一注册方法：
from fastapi import FastAPIfrom .usetime_middleware import UseTimeMiddlewaredef registerMiddlewareHandle(server: FastAPI): # 添加耗时请求中间件 server.add_middleware(UseTimeMiddleware)
2.3 调用注册函数
修改main.py文件，修改内容如下:
from app import errors, middleware...# 实例化server = FastAPI(redoc_url=None, docs_url="/apidoc", title="FastAPI学习")# # 注册中间件middleware.registerMiddlewareHandle(server)...
2.4 添加路由
修改app/router/demo_router.py文件，新增内容如下:
@router.get("/middle/useTime")async def middleUseTime() -> response.HttpResponse: """ 中间件使用-演示 """ # 随机暂停时间 seconds = random.randint(500, 5000) / 1000 print("暂停时间:", seconds) time.sleep(seconds) return response.ResponseSuccess(seconds)
2.5 验证
3.多中间件顺序
3.1 创建多个中间件
# 文件: app/middleware/test_middleware.pyfrom fastapi import Requestfrom starlette.middleware.base import BaseHTTPMiddlewarefrom starlette.responses import Responseclass TestMiddleware(BaseHTTPMiddleware): """ 测试顺序中间件""" def __init__(self, app): super().__init__(app) async def dispatch(self, request: Request, call_next) -> Response: print("调用-中间件-TestMiddleware---before") # 调用下一个中间件或路由处理函数 result = await call_next(request) print("调用-中间件-TestMiddleware---after") return result # -------------- 另外一个中间件 --------------# 文件: app/middleware/test_middleware.pyfrom fastapi import Requestfrom starlette.middleware.base import BaseHTTPMiddlewarefrom starlette.responses import Responseclass TokenMiddleware(BaseHTTPMiddleware): """ token验证中间件 """ def __init__(self, app): super().__init__(app) async def dispatch(self, request: Request, call_next) -> Response: token = request.headers.get("X-Token", "") print("调用-token验证中间件-TokenMiddleware---before", token) result = await call_next(request) print("调用-token验证中间件-TokenMiddleware---after", token) return result
3.2 注册
修文件app/middleware/__init__.py中，统一注册中间件方法：
from fastapi import FastAPIfrom .usetime_middleware import UseTimeMiddlewarefrom .token_middleware import TokenMiddlewarefrom .test_middleware import TestMiddlewaredef registerMiddlewareHandle(server: FastAPI): # 添加token验证中间件 server.add_middleware(TokenMiddleware) # 添加耗时请求中间件 server.add_middleware(UseTimeMiddleware) # 测试 server.add_middleware(TestMiddleware)
3.3 请求验证
@总结: 通过实践发现: 中间件的执行顺序和注册顺序，正好是相反的；先注册的后执行，
3.4 add_middleware函数
def add_middleware(self, middleware_class: type, **options: typing.Any) -> None: if self.middleware_stack is not None: # pragma: no cover raise RuntimeError("Cannot add middleware after an application has started") # 发现每次调用，都会把最后一个中间件放在 索引=0的位置，即最前面 self.user_middleware.insert(0, Middleware(middleware_class, **options))
3.5 优化注册函数
add_middleware函数通过栈的方式(后进先出)注册中间件，跟我们的思维相反，如果注册的中间件多的话，排查问题时，很容易被惯性思维误导，下面优化了下注册函数: 先注册的先执行
from fastapi import FastAPIfrom .usetime_middleware import UseTimeMiddlewarefrom .token_middleware import TokenMiddlewarefrom .test_middleware import TestMiddleware# 定义注册顺序middlewareList = [ UseTimeMiddleware, # 添加耗时请求中间件 TokenMiddleware, # 添加token验证中间件 TestMiddleware # 测试中间件]def registerMiddlewareHandle(server: FastAPI): # 倒序中间件 middlewareList.reverse() # 遍历注册 for _middleware in middlewareList: server.add_middleware(_middleware)
4.内置中间件
4.1 常用中间件
框架也提供了一些常用的的内置中间件
HTTPSRedirectMiddleware: 将
HTTP 请求重定向到
HTTPS。这个中间件会检查请求的协议，如果是
HTTP，则自动将请求重定向到相应的
HTTPS 地址;
TrustedHostMiddleware: 强制所有传入请求都具有正确设置的
Host 标头，以防止 HTTP 主机标头攻击。
GZipMiddleware: 用于在响应中压缩内容，以减小传输大小。这有助于提高应用程序的性能，特别是在处理大量文本或数据时。
CORSMiddleware: 用于处理跨域资源共享（CORS）请求。CORS 是一种浏览器机制，允许 Web 页面从不同的域请求不同域的资源。
4.2 CORSMiddleware使用
跨域中间件应该是我们常用的一种中间件，具体使用示例如下:
# 导入from fastapi.middleware.cors import CORSMiddleware# 注册 server.add_middleware( CORSMiddleware, allow_origins=["*"], # 允许的来源，可以是字符串、字符串列表，或通配符 "*" allow_credentials=True, # 是否允许携带凭证（例如，使用 HTTP 认证、Cookie 等） allow_methods=["*"], # 允许的 HTTP 方法，可以是字符串、字符串列表，或通配符 "*" allow_headers=["*"], # 允许的 HTTP 头信息，可以是字符串、字符串列表，或通配符 "*" expose_headers=["*"], # 允许前端访问的额外响应头，可以是字符串、字符串列表 max_age=600, # 请求的缓存时间，以秒为单位 )
以下是常用参数的详细说明：
allow_origins: 允许的来源。可以是字符串、字符串列表，或通配符
"*" 表示允许所有来源。
allow_credentials: 是否允许携带凭证（例如，使用
HTTP 认证、
Cookie 等）。默认为
False,如果为
True，
allow_origins 必须为具体的源，不可以是
["*"]。
allow_methods: 允许的 HTTP 方法，可以是字符串、字符串列表，或通配符
"*" 表示允许所有方法。
allow_headers: 允许的 HTTP 头信息，可以是字符串、字符串列表，或通配符
"*" 表示允许所有头信息。
expose_headers: 允许前端访问的额外响应头，可以是字符串、字符串列表,一般很少指定。
max_age: 浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600,一般使用默认值。
4.3 GZipMiddleware使用
# 导入from fastapi.middleware.gzip import GZipMiddleware# 注册server.add_middleware( GZipMiddleware, minimum_size=500, # 启用 Gzip 压缩的最小响应体大小，单位为字节 compress_level=6, # Gzip 压缩级别，范围为 0 到 9，级别越高，压缩率越高，但耗费的 CPU 越多 exclude_mediatypes=["application/json"], # 不进行 Gzip 压缩的媒体类型，可以是字符串或字符串列表)
以下是常用参数的详细说明：
minimum_size: 启用
Gzip 压缩的最小响应体大小，单位为字节。只有当响应体的内容大于等于
minimum_size 时，才会进行
Gzip 压缩。默认为
500 字节。
compress_level:
Gzip 压缩级别，范围为 0 到 9。级别越高，压缩率越高，但耗费的 CPU 越多。默认为
6。
exclude_mediatypes: 不进行
Gzip 压缩的媒体类型，可以是字符串或字符串列表。例如，可以配置不对
JSON 响应进行压缩。
5.JWT验证
FastAPI框架没有内置的 JWT（JSON Web Token）中间件，我们可以使用第三方库（ PyJWT）来验证和解码 JWT，下面来实现使用示例；
5.1 安装
$ pip install pyjwt
5.2 生成token
# 定义秘钥和算法secret_key = "abcd12345@abcdef"algorithm = "HS256"def jwtGenerator() -> str: # 构造 payload payload = { "uid": 1234567, # 主题，通常是用户的唯一标识 "iat": datetime.utcnow(), # 签发时间 "exp": datetime.utcnow() + timedelta(minutes=30), # 过期时间 "data": {"user_name": "张三", "uid": 1234567, "phone": "17600000000"} # 自定义的数据 } # 生成 JWT return jwt.encode(payload, secret_key, algorithm=algorithm) if __name__ == '__main__': # 生成token token = jwtGenerator() print("Token:", token) """Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1aWQiOjEyMzQ1NjcsImlhdCI6MTcwMjUyODA1MSwiZXhwIjoxNzAyNTI5ODUxLCJkYXRhIjp7InVzZXJfbmFtZSI6Ilx1NWYyMFx1NGUwOSIsInVpZCI6MTIzNDU2NywicGhvbmUiOiIxNzYwMDAwMDAwMCJ9fQ.sV_k75YPdEtI2P7-HlDbGbXWdcdosf1cCImNkux7OGg"""
5.3 校验token
def parseToken(jwtToken: str) -> str: """ 解析token """ try: return jwt.decode(jwtToken, secret_key, algorithms=[algorithm]) except jwt.ExpiredSignatureError: print("JWT has expired.") return "" except jwt.InvalidTokenError: print("Invalid JWT.") return "" if __name__ == '__main__': token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1aWQiOjEyMzQ1NjcsImlhdCI6MTcwMjUyODA1MSwiZXhwIjoxNzAyNTI5ODUxLCJkYXRhIjp7InVzZXJfbmFtZSI6Ilx1NWYyMFx1NGUwOSIsInVpZCI6MTIzNDU2NywicGhvbmUiOiIxNzYwMDAwMDAwMCJ9fQ.sV_k75YPdEtI2P7-HlDbGbXWdcdosf1cCImNkux7OGg" # 验证token parseToken = parseToken(token) print("parseToken:", parseToken) """parseToken: {'uid': 1234567, 'iat': 1702528051, 'exp': 1702529851, 'data': {'user_name': '张三', 'uid': 1234567, 'phone': '17600000000'}}"""
5.4 封装中间件
在包app/middleware下，并新增文件jwt_middleware.py，文件内容如下:
from fastapi import Request, statusfrom fastapi.encoders import jsonable_encoderfrom fastapi.responses import JSONResponsefrom starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpointfrom starlette.responses import Responsefrom app.types import response, JwtDatafrom app.utils import JwtManageUtil# 后期做配置，这里临时演示secret_key = "abcd12345@abcdef"# 不检查noCheckTokenPathList = [ "/apidoc", "/openapi.json", "/api/user/login"]class JwtMiddleware(BaseHTTPMiddleware): """ jwt验证中间件 """ def __init__(self, app): super().__init__(app) self.jwtUtil = JwtManageUtil(secretKey=secret_key) async def dispatch(self, request: Request, call_next): # 判断路由是否需要验证 path = request.url.path if path in noCheckTokenPathList: return await call_next(request) # 获取token token = request.headers.get('x-token', '') if token == "": return JSONResponse( status_code=status.HTTP_200_OK, content=jsonable_encoder(response.ResponseFail('token不能为空~'))) # 验证token tokenInfo = self.jwtUtil.decode(token, JwtData) if not isinstance(tokenInfo, JwtData): # 验证失败 return JSONResponse( status_code=status.HTTP_200_OK, content=jsonable_encoder(response.ResponseFail(tokenInfo))) result = await call_next(request) print("token解析成功", tokenInfo) return result
限于文章篇幅，上面示例中JwtManageUtil代码不在展示，具体源代码可在 微信搜索【猿码记】回复 【fastapi】获取
5.5 请求验证
$ curl -X 'GET' \ 'http://127.0.0.1:8000/demo/middle/useTime' \ -H 'accept: application/json' \ -H 'X-Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiIyMDIzMTIxNDIwMzEyNzk2ODUiLCJpc3MiOiJcdTczM2ZcdTc4MDFcdThiYjAiLCJpYXQiOjE3MDI1NTcxMDYsImV4cCI6MTcwMjU2MDcwNiwiZGF0YSI6eyJ1aWQiOjExMjIzMzQ0LCJ1bmFtZSI6Ilx1NWYyMFx1NGUwOSJ9fQ.62Nijbs08Oy1wo1-IBfO9RvTGQ3B6aGtaAQt0qrT7-4'# 返回{"code":200,"msg":"处理成功","data":1.096,"additional":{"time":"2023-12-14 20:35:00","trace_id":"90060a47fa9850d1ccb1bc9fc1045c8c"}}# 故意写错token$ curl -X 'GET' \ 'http://127.0.0.1:8000/demo/middle/useTime' \ -H 'accept: application/json' \ -H 'X-Token: 12334'# 返回{"code":-1,"msg":"TokenInvalid|token非法","data":null,"additional":{"time":"2023-12-14 20:35:26","trace_id":"5303bc01350d7ab7f029098378d6a264"}}
本文由 mdnice 多平台发布
好文推荐
发表评论