Fastapi 自定义中间件
这里的中间件主要作用其实就是在请求前和请求后处理机制。通常我们的可以在中间件里处理的事情有：
日志记录
鉴权
数据库的操作开关处理等
如在之前的Flask中其实是比较简单的，如flask中几个钩子的函数：
- before_first_request：第一个请求运行前
- before_request：每次请求前运行。
- after_request：处理逻辑没有异常抛出，每次请求后运行（这里可以返回我们的自定义的响应体）
- teardown_request：在每次请求后运行，即使处理发生了错误。
- teardown_appcontext：在应用上下文从栈中弹出之前运行
复制代码
但是在fastapi中的处理没有类似上述的函数的回调。它的中间件类似的我们GO语言的gin中间件一样。以下是我自己对中间件使用过程中遇到的一些问题的整理。
1：最简的http请求类型中间件
如官网提供的最简自定义的中间件示例：
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from starlette.responses import StreamingResponse, JSONResponse
import json
app = FastAPI()
@app.middleware("http")
async def log_request(request, call_next):
print('请求开始前我可以处理事情11111')
response = await call_next(request)
print('请求开始后我可以处理的事情33333333333')
return response
@app.get("/")
async def not_timed():
print('请求开始后我可以处理的事情222222')
return {"message": "你好"}
import uvicorn
if __name__ == '__main__':
# 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
uvicorn.run('main:app', host="127.0.0.1", port=8000, debug=True, reload=True)
复制代码
输出结果为：
请求开始前我可以处理事情11111
请求开始后我可以处理的事情222222
请求开始后我可以处理的事情33333333333
INFO: 127.0.0.1:1084 - "GET HTTP/1.1" 200 OK
复制代码
2：基于BaseHTTPMiddleware的实现的中间件
多自定义中间件示例：
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from starlette.responses import StreamingResponse, JSONResponse
import json
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
@app.middleware("http")
async def log_request(request, call_next):
print('请求开始前我可以处理事情11111')
response = await call_next(request)
print('请求开始后我可以处理的事情33333333333')
return response
# 基于BaseHTTPMiddleware的中间件实例，
class CostimeHeaderMiddleware(BaseHTTPMiddleware):
# dispatch 必须实现
async def dispatch(self, request, call_next):
print('请求开始前我可以处理事情44444444444444')
start_time = time.time()
responser = await call_next(request)
process_time = round(time.time() - start_time, 4)
# 返回接口响应时间
responser.headers["X-Process-Time"] = f"{process_time} (s)"
print('请求开始前我可以处理事情555555555')
return responser
# 基于BaseHTTPMiddleware的中间件实例，
class CostimeHeaderMiddleware2(BaseHTTPMiddleware):
# dispatch 必须实现
async def dispatch(self, request, call_next):
print('请求开始前我可以处理事情666666666666666666666')
start_time = time.time()
responser = await call_next(request)
process_time = round(time.time() - start_time, 4)
# 返回接口响应时间
responser.headers["X-Process-Time"] = f"{process_time} (s)"
print('请求开始前我可以处理事情7777777777777777777777')
return responser
app.add_middleware(CostimeHeaderMiddleware)
app.add_middleware(CostimeHeaderMiddleware2)
@app.get("/")
async def not_timed():
print('请求开始后我可以处理的事情222222')
return {"message": "你好"}
import uvicorn
if __name__ == '__main__':
# 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
uvicorn.run('main2:app', host="127.0.0.1", port=8000, debug=True, reload=True)
复制代码
输出结果为：
请求开始前我可以处理事情666666666666666666666
请求开始前我可以处理事情44444444444444
请求开始前我可以处理事情11111
请求开始后我可以处理的事情222222
请求开始后我可以处理的事情33333333333
请求开始前我可以处理事情555555555
请求开始前我可以处理事情7777777777777777777777
INFO: 127.0.0.1:1998 - "GET HTTP/1.1" 200 OK
复制代码
PS1：从上面的输出结果可以看得到我们的中间件的注册顺序非常的重要。他们的上面的注册顺序是：
---->越是最晚注册的，它就会在中间件的洋葱模型的最外层。
---->越是最早注册的，它就会在中间件的洋葱模型的最内层。
PS2：假如你有需要使用中间件来处理全局异常的捕获的话，则当然是放在最外层去处理咯！
3：中间件中获取最终responser返回值
通常有上面的需求的话，一般是放在我们的日志记录中，请求完成后，我们的日志记录我们返回给客户端的东西，这时候，就需要在中间里获取到我们的最终的响应体的报文内容。
但是是打印出来看我们的responser，你会发现它是一个：
<starlette.responses.StreamingResponse object at 0x0000020C287EABA8>
打印 print(response.dict)：你会打得到下面的信息：
{'body_iterator': <async_generator object BaseHTTPMiddleware.call_next.<locals>.body_stream at 0x0000020C2886B6A8>, 'status_code': 200, 'media_type': None, 'background': None, 'raw_headers': [(b'content-length', b'20'), (b'content-type', b'application/json')]}
复制代码
我们发现好像根本无法直接的获取到我们的响应报文信息，从上面可以看得出我们的响应报文应该是在async_generator object BaseHTTPMiddleware这个异步生成器里面了！
翻看官方提的issues,也有比人遇到类似的需求询问
[QUESTION] How to get Response Body from middleware #954
https://github.com/tiangolo/fastapi/issues/954
复制代码
不过翻看了一下，最终没有一个满意的答案。
然后不经意间在stackoverflow.com 翻看到另一个关于这个的需求的实现，有一个老铁是已经给出了就具体方案：具体地址为：stackoverflow.com/questions/6… 已下是完整的示例：
#!/usr/bin/evn python
# coding=utf-8
# + + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + +
# ┏┓ ┏┓+ +
# ┏┛┻━━━┛┻┓ + +
# ┃ ┃
# ┃ ━ ┃ ++ + + +
# ████━████ ┃+
# ┃ ┃ +
# ┃ ┻ ┃
# ┃ ┃ + +
# ┗━┓ ┏━┛
# ┃ ┃
# ┃ ┃ + + + +
# ┃ ┃ Codes are far away from bugs with the animal protecting
# ┃ ┃ + 神兽保佑,代码无bug
# ┃ ┃
# ┃ ┃ +
# ┃ ┗━━━┓ + +
# ┃ ┣┓
# ┃ ┏┛
# ┗┓┓┏━┳┓┏┛ + + + +
# ┃┫┫ ┃┫┫
# ┗┻┛ ┗┻┛+ + + +
# + + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + ++ + + +"""
"""
Author = zyx
@version: v1.0.0
@File: __init__.py.py
@文件功能描述:------
"""
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from starlette.responses import StreamingResponse, JSONResponse
import json
app = FastAPI()
class aiwrap:
def __init__(self, obj):
self._it = iter(obj)
def __aiter__(self):
return self
async def __anext__(self):
try:
value = next(self._it)
except StopIteration:
raise StopAsyncIteration
return value
@app.middleware("http")
async def log_request(request, call_next):
print('请求开始前我可以处理事情11111')
response = await call_next(request)
print('请求开始后我可以处理的事情33333333333',response)
resp_body = [section async for section in response.__dict__['body_iterator']]
# Repairing FastAPI response
response.__setattr__('body_iterator', aiwrap(resp_body))
# Formatting response body for logging
try:
resp_body = json.loads(resp_body[0].decode())
except:
resp_body = str(resp_body)
print("中间件里面获取到最终返回的响应体的信息", resp_body)
print(response.__dict__)
return response
@app.get("/")
async def not_timed():
return {"message": "Not timed"}
import uvicorn
if __name__ == '__main__':
# 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
uvicorn.run('main:app', host="127.0.0.1", port=8000, debug=True, reload=True)
复制代码
输出结果是：
请求开始前我可以处理事情11111
请求开始后我可以处理的事情33333333333 <starlette.responses.StreamingResponse object at 0x0000026BA8369860>
中间件里面获取到最终返回的响应体的信息 {'message': 'Not timed'}
{'body_iterator': <main.aiwrap object at 0x0000026BA846B780>, 'status_code': 200, 'media_type': None, 'background': None, 'raw_headers': [(b'content-length', b'23'), (b'content-type', b'application/json')]}
INFO: 127.0.0.1:5674 - "GET HTTP/1.1" 200 OK
复制代码
4：取巧方式,通过 req.state.来写入响应报文，然后在中间里读取
因为我们的响应的报文是在中间里面内层执行的，所以我们的可以对我们的请求体里面写入我们的响应报文即可，不过这种方式可能需要在我们的ApiResponse(JSONResponse)里面进行对于的Request进行传递才可以仅供参考！
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from starlette.responses import StreamingResponse, JSONResponse
import json
from typing import Optional,Dict,Any
app = FastAPI()
@app.middleware("http")
async def log_request(request: Request, call_next):
print('请求开始前我可以处理事情11111')
response = await call_next(request)
print('请求开始后我可以处理的事情33333333333', response)
# Repairing FastAPI response
print("另一种取巧的方式：中间件里面获取到最终返回的响应体的信息", request.state.rspbody)
print(response.__dict__)
return response
class ApiResponse(JSONResponse):
# 定义返回响应码--如果不指定的话则默认都是返回200
http_status_code = 200
# 默认成功
code = 0
# 默认Node.如果是必选的，去掉默认值即可
data: Optional[Dict[str, Any]] = None # 结果可以是{} 或 []
msg = '成功'
def __init__(self, req: Request = None, http_status_code=None, code=None, data=None, msg=None, **options):
if data:
self.data = data
if msg:
self.msg = msg
if code:
self.code = code
if http_status_code:
self.http_status_code = http_status_code
# 返回内容体
body = dict(
msg=self.msg,
code=self.code,
data=self.data,
)
if req:
pass
req.state.rspbody = body
super(ApiResponse, self).__init__(status_code=self.http_status_code, content=body, **options)
@app.get("/")
async def not_timed(req: Request):
return ApiResponse(req=req)
import uvicorn
if __name__ == '__main__':
# 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：uvicorn xxx:app --reload
uvicorn.run('main:app', host="127.0.0.1", port=8000, debug=True, reload=True)
复制代码
输出的结果为：
请求开始前我可以处理事情11111
请求开始后我可以处理的事情33333333333 <starlette.responses.StreamingResponse object at 0x0000029C640835C0>
另一种取巧的方式：中间件里面获取到最终返回的响应体的信息 {'msg': '成功', 'code': 0, 'data': None}
{'body_iterator': <async_generator object BaseHTTPMiddleware.call_next.<locals>.body_stream at 0x0000029C6407F598>, 'status_code': 200, 'media_type': None, 'background': None, 'raw_headers': [(b'content-length', b'37'), (b'content-type', b'application/json')]}
INFO: 127.0.0.1:1307 - "GET / HTTP/1.1" 200 OK
复制代码
结尾
简单小笔记！仅供参考！
END
简书：www.jianshu.com/u/d6960089b…
掘金：juejin.cn/user/296393…
公众号：微信搜【小儿来一壶枸杞酒泡茶】
小钟同学 | 文 【原创】【欢迎一起学习交流】| QQ：308711822
文章转载自小儿来一壶枸杞酒泡茶，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
