【FastAPI】中间件（Middleware）开发

什么是中间件

中间件是一个函数

请求达到 响应逻辑之前 会经过一层或多层 中间件 ，响应结果返回客户端也会经过一层或多层 中间件

FastAPI 中间件开发

自定义中间件

import time

import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import Response

app = FastAPI()

@app.middleware('http') # 中间件类型 现在只支持 http 类型
async def add_process_time_header(request: Request, call_next):
    """
    计算请求响应时间
    :param request: 请求
    :param call_next: 请求处理回调
    :return:
    """
    start_time = time.time()
    res: Response = await call_next(request)
    process_time = time.time() - start_time
    res.headers['X-Process-Time'] = str(process_time)
    return res

@app.get('/middleware')
def middleware():
    return '没有处理逻辑，响应成功'

if __name__ == '__main__':
    uvicorn.run(app)

提供的中间件

CORSMiddleware

跨域资源共享中间件

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=[  # 允许访问的域 域 --> 协议:域名:端口
        'http://127.0.0.1',
        'http://127.0.0.1:8000',
    ],
    allow_methods=['*', 'GET'],  # * 可以通配
    allow_headers=['*'],  # 头部
    allow_credentials=False,  # HTTPS 证书
    allow_origin_regex=None,  # 正则表达式匹配  'https://.*\.example\.org
    expose_headers=[],  # 指明可以被 浏览器访问 的 响应头
    max_age=600  # 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

HTTPSRedirectMiddleware

强制所有传入请求必须是https或wss

如果不是 https 则会自动跳转到 https，如果网站没有配置 https ，则报错

h11._util.RemoteProtocolError: illegal request line 
WARNING:  Invalid HTTP request received.

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

app.add_middleware(HTTPSRedirectMiddleware)

@app.get("/")
async def main():
    return {"message": "Hello World"}

TrustedHostMiddleware

强制所有传入请求都具有正确设置的Host标头，以防止 HTTP 主机标头攻击。

如果不包含，则响应 400 状态码，并返回 Invalid host header

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

app.add_middleware(
    TrustedHostMiddleware, 
    allowed_hosts=["bigdataboy.cn", "*.bigdataboy.cn"] # 允许的 hosts 列表
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

GZipMiddleware

处理包含”gzip”在Accept-Encoding标头中的任何请求的 GZip 响应。

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

app.add_middleware(
    GZipMiddleware,
    minimum_size=1000)  # 不要 GZip 响应小于此最小字节大小。默认为500.

@app.get("/")
async def main():
    return {"message": "Hello World"}

带 yield 关键字依赖, 依赖中的退出代码将在执行中间件后执行

def get_db():
    db = dbSession()
    try:
        yield db
    finally:
        db.close()