目 录CONTENT

文章目录

FastApi请求拦截

phyger
2022-02-16 / 0 评论 / 1 点赞 / 1,184 阅读 / 3,070 字 / 正在检测是否收录...

前言

我们经常听说请求拦截,那到底什么是请求拦截,请求拦截有什么用呢?今天我们就一起来看一看。

关于请求拦截

请求拦截,顾名思义就是在请求过程中将请求拦截下来,然后对请求进行处理然后才进入视图中处理然后响应给客户端。

在安全测试、前后端开发中,请求拦截是非常有用的。比如 token 续签、统一响应处理、统一异常处理、历史接口改造等。

今天我们就用非常简单的 FastApi 请求拦截例子来深入理解请求拦截。

原始代码

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI()

from random import SystemRandom

SALT_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

def gen_str(length):
    sys_random = SystemRandom()
    return "".join(sys_random.choice(SALT_CHARS) for i in range(length))

@app.get('/')
def index():
    return 'Welcome to Python研究所!'

@app.get('/str')
def genStr(request: Request,num:int):
    return gen_str(num)

if __name__=='__main__':
    uvicorn.run(app='demo1:app',host='localhost',port=1213,reload=True)

如上代码有两个接口,一个是/根接口,还有一个是根据参数生成指定长度字符串的接口。

原始str接口的效果

原始代码的response headers内容

需求 1

假设我们现在需要向 response 中增加一个参数来告诉客户端我们这个请求处理所花费的时间,我们可以使用 FastApi 的中间件来实现。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn,time

app = FastAPI()

# 为app增加接口处理耗时的响应头信息
@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

    # X- 作为前缀代表专有自定义请求头
    response.headers["X-Process-Time"] = str(process_time)
    return response

from random import SystemRandom

SALT_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

def gen_str(length):
    sys_random = SystemRandom()
    return "".join(sys_random.choice(SALT_CHARS) for i in range(length))

@app.get('/')
def index():
    return 'Welcome to Python研究所!'

@app.get('/str')
def genStr(request: Request,num:int):
    return gen_str(num)

if __name__=='__main__':
    uvicorn.run(app='demo1:app',host='localhost',port=1213,reload=True)

如上,我们直接给 app 增加了 http 的中间件,在其中捕捉了请求,记录时间,然后发起请求,在响应之前给 response 增加了响应头信息 X-Process-Time,然后将新的响应返回给客户端。

处理后的效果

需求 2

我们再演示一个很常见的场景,就是当我们发现客户端的 token 快过期的时候我们为其自动续签。

以上面的代码为例,我们做一个全局请求拦截器,如果请求头携带的 token 快过期了,我们就自动刷新 token

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse,RedirectResponse
import uvicorn,time

app = FastAPI()

# 为app增加接口处理耗时的响应头信息
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    hs = request.headers
    token=int(hs.get("token"))
    print(token,type(token))
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time

    # X- 作为前缀代表专有自定义请求头
    response.headers["X-Process-Time"] = str(process_time)

    if token:
        if token<=5:
            response.headers["X-token"] = str(10)
    return response

from random import SystemRandom

SALT_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

def gen_str(length):
    sys_random = SystemRandom()
    return "".join(sys_random.choice(SALT_CHARS) for i in range(length))

@app.get('/')
def index():
    return 'Welcome to Python研究所!'

@app.get('/str')
def genStr(request: Request,num:int):
    return gen_str(num)

if __name__=='__main__':
    uvicorn.run(app='demo1:app',host='localhost',port=1213,reload=True)

如上,我们在请求拦截器中增加了对请求头中国 token 值的判断,为了方便演示,我们直接将 token 定义为数字。token 小于 5 的时候,我们就对 token 进行自动续签为 10。如果 token 大于 5,我们就正常进行请求处理。

token有效的情况

token快失效的情况

如上,对于快到期的 token,请求拦截器已经自动进行了 token 刷新,前端收到新 token 后对客户端的 token 进行覆盖即可。

以上就是今天的全部内容那个了,希望能够对你有所帮助。

1

评论区