1.介绍
1.1 官网介绍
"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.
-
它接收你的应用程序的每一个 请求. -
然后它可以对这个 请求做一些事情或者执行任何需要的代码. -
然后它将 请求传递给应用程序的其他部分 (通过某种 路径操作). -
然后它获取应用程序生产的 响应 (通过某种 路径操作). -
它可以对该 响应做些什么或者执行任何需要的代码. -
然后它返回这个 响应.
1.2 中间件工作示意图
1.3 官方使用示例
import time
from fastapi import FastAPI, Request
app = 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 time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
class 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 FastAPI
from .usetime_middleware import UseTimeMiddleware
def 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.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
class 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.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
class 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 FastAPI
from .usetime_middleware import UseTimeMiddleware
from .token_middleware import TokenMiddleware
from .test_middleware import TestMiddleware
def 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 FastAPI
from .usetime_middleware import UseTimeMiddleware
from .token_middleware import TokenMiddleware
from .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, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response
from app.types import response, JwtData
from 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 多平台发布