1. 中间件
1.1 介绍(ChatGPT抄的,大致可以理解)
一种机制,用于在处理请求和响应之前对其进行拦截、处理或修改。中间件可以在应用程序的请求处理管道中插入自定义逻辑,以实现一些通用的功能,如身份验证、日志记录、错误处理等。
中间件在请求到达应用程序之前被执行,并在应用程序生成响应之前被执行。它们可以对请求进行修改、记录请求信息、添加额外的响应头,甚至可以拒绝请求或提前返回响应。
在FastAPI中,可以通过编写中间件函数来定义中间件逻辑。这些函数接收一个Request对象和一个Callable对象(代表下一个中间件或请求处理器)作为参数,并返回一个Response对象。
注释:
我理解就是在中间件里写一个函数,所有路由函数在执行前都会执行此函数
1.2 示例
import uvicorn
from fastapi import FastAPI, Request
app = FastAPI()
# 中间件
@app.middleware("http")
async def m2(request:Request, call_next):
# 请求代码块
print("m2 request")
response = await call_next(request)
# 响应代码块
print("m2 response")
return response
@app.middleware("http")
async def m1(request: Request, call_next):
# 请求代码块
print("m1 request")
response = await call_next(request)
# 响应代码块
print("m1 response")
return response
@app.get("/user")
async def get_user():
print("get_user函数执行")
return {"user":"current user"}
@app.get("/item/{item_id}")
async def get_item(item_id:int):
print("get_item函数执行")
return {"item_id":item_id}
if __name__ == "__main__":
uvicorn.run("main:app", port=5050, reload=True)
注释:
执行顺序为:request先下后上,response先上后下,先执行request,再执行response
1.3 示例
import time
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import Response
app = FastAPI()
# 中间件
@app.middleware("http")
async def m2(request:Request, call_next):
# 请求代码块
print("m2 request")
response = await call_next(request)
# 响应代码块
print("m2 response")
return response
@app.middleware("http")
async def m1(request: Request, call_next):
# 请求代码块
print("m1 request")
# 示例:访问url限制
if request.client.host in ["127.0.0.1",]:
return Response(status_code=403, content="visit forbidden")
# 示例:访问路径限制
if request.url.path in ["/user"]:
return Response(status_code=403, content="visit forbidden")
response = await call_next(request)
# 响应代码块
print("m1 response")
return response
@app.get("/user")
async def get_user():
print("get_user函数执行")
return {"user":"current user"}
@app.get("/item/{item_id}")
async def get_item(item_id:int):
print("get_item函数执行")
return {"item_id":item_id}
if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=5050, reload=True)
1.4 response响应示例
import time
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import Response
app = FastAPI()
# 中间件
@app.middleware("http")
async def m2(request:Request, call_next):
# 请求代码块
print("m2 request")
response = await call_next(request)
# 响应代码块
print("m2 response")
return response
@app.middleware("http")
async def m1(request: Request, call_next):
# 请求代码块
print("m1 request")
# 示例:访问url限制
# if request.client.host in ["127.0.0.1",]:
# return Response(status_code=403, content="visit forbidden")
# # 示例:访问路径限制
# if request.url.path in ["/user"]:
# return Response(status_code=403, content="visit forbidden")
start = time.time()
response = await call_next(request)
# 响应代码块
print("m1 response")
end = time.time()
response.headers["ProcessTimer"] = str(end - start)
return response
@app.get("/user")
async def get_user():
time.sleep(3)
print("get_user函数执行")
return {"user":"current user"}
@app.get("/item/{item_id}")
async def get_item(item_id:int):
time.sleep(2)
print("get_item函数执行")
return {"item_id":item_id}
if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=5050, reload=True)
2. CORS跨域请求
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_methods=['GET', 'POST'],
allow_headers=['*']
)
# @app.middleware("http")
# async def HelloMiddleware(request:Request, call_next):
# response = await call_next(request)
# response.headers["Access-Control-Allow-Origin"] = "*"
# return response
@app.get("/hello")
async def get_hello():
return {"message": "hello xujie"}
if __name__ == "__main__":
uvicorn.run("main:app", port=5050, reload=True)
注释:
fastapi框架(测试开发用)差不多就这些了,请求、响应、assert、数据库、中间件、跨域、template、基本上合作开发一套简单自用测试平台已经够用(至少我是这样的),填坑完成,撒花