FastApi请求拦截

Python研究所

发布于 2022-06-17 09:09:24

2.4K0

发布于 2022-06-17 09:09:24

文章被收录于专栏：大飞的部落阁

前言

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

关于请求拦截

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

在安全测试、前后端开发中，请求拦截是非常有用的。比如 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)

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

需求 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 进行覆盖即可。

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

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

网络安全

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

网络安全

登录后参与评论

0 条评论

热度