FastApi的请求拦截-阿里云开发者社区

FastApi的请求拦截

2022-06-19 875

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： FastApi的请求拦截

前言

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

关于请求拦截

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

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