“中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

  • 它接收你的应用程序的每一个请求.
  • 然后它可以对这个请求做一些事情或者执行任何需要的代码.
  • 然后它将请求传递给应用程序的其他部分 (通过某种路径操作).
  • 然后它获取应用程序生产的响应 (通过某种路径操作).
  • 它可以对该响应做些什么或者执行任何需要的代码.
  • 然后它返回这个 响应.

创建中间件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import time

from fastapi import FastAPI, Request

app = FastAPI()


@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
    response.headers["X-Process-Time"] = str(process_time)
    return response

封装中间件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class UseTimeMiddleware(BaseHTTPMiddleware):
    """ 计算耗时中间件"""

    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next) -> Response:
        """ 请求耗时 """
        start_time = time.time()
        # 调用下一个中间件或路由处理函数
        result = await call_next(request)
        process_time = time.time() - start_time
        result.headers["X-Process-Time"] = str(process_time)
        return result

定义的中间件类UseTimeMiddleware要继承基础类BaseHTTPMiddleware

自定义安全中间件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp, Scope

class CustomHeadersMiddleware(BaseHTTPMiddleware):
    """ 设置自定义安全标头中间件 """

    def __init__(self, app: ASGIApp, headers: dict):
        super().__init__(app)
        self.headers = headers

    async def dispatch(self, request: Scope, call_next):
        response = await call_next(request)
        for header, value in self.headers.items():
            response.headers[header] = value
        return response

custom_headers = {
        'X-DNS-Prefetch-Control': 'off',
        'X-Download-Options': 'noopen',
        'X-Permitted-Cross-Domain-Policies': 'none',
    }
app.add_middleware(
    CustomHeadersMiddleware,
    headers=custom_headers,
)

参考