Python FastAPI系列：常用FastAPI middleware中间件的详细使用_fastapi sessionmiddleware

在FastAPI中已经内置了Starlette大量的middleware中间件，这些中间件可以初步用于OAuth2、CORS、Gzip等功能的实现。下面我们对这些常用中间件的使用进行详细的说明。

多个中间件的执行顺序

在一个FastAPI程序中，可以通过add_middleware引入多个中间件，这时就会出现执行顺序的问题。FastAPI遵循后进先执行的原则，参考下面的例子：
main.py

import uvicorn as uvicorn
from fastapi import FastAPI

from fapi.middlewares.multi_middleware import MutilAMiddleware, MutilBMiddleware

app = FastAPI()

app.add_middleware(MutilAMiddleware, assigned_number=1)
app.add_middleware(MutilBMiddleware, assigned_number=2)


@app.get("/")
async def index():
    print("'index' router executed!")
    return "test middleware"


if __name__ == '__main__':
    uvicorn.run(app="main:app", port=8088, reload=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

multi_middleware.py

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request


class MutilAMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, assigned_number):
        super().__init__(app)
        self.assigned_number = assigned_number

    async def dispatch(self, request: Request, call_next):
        print(f"{self.assigned_number}:'A' middleware executed!")
        response = await call_next(request)
        return response


class MutilBMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, assigned_number):
        super().__init__(app)
        self.assigned_number = assigned_number

    async def dispatch(self, request: Request, call_next):
        print(f"{self.assigned_number}:'B' middleware executed!")
        response = await call_next(request)
        return response
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

以上代码的执行结果为：

使用已有的中间件

FastAPI内置中间件

FastAPI借用了Starlette提供的内置中间件，包括：
SessionMiddleware：提供了基于cookie的HTTP会话管理功能。

import uvicorn as uvicorn
from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from starlette.requests import Request

app = FastAPI()

app.add_middleware(SessionMiddleware, secret_key="middleware_secret_key")


@app.get("/set_session")
async def set_session(request: Request):
    request.session["session_data"] = "some session data"
    return 'set session data success!'


@app.get("/read_session")
async def read_session(request: Request):
    return f'session data is: {request.session.get("session_data")}'


if __name__ == '__main__':
    uvicorn.run(app="main:app", port=8088, reload=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

SessionMiddleware中间件有如下配置参数：
secret_key - 随机字符串作为secret_key
session_cookie - cookie名称，默认‘session’
max_age - 会话过期时间，以秒为单位。默认为2周。如果设置为None，则cookie将持续存在，直到浏览器会话结束。
same_site - 标志阻止浏览器在跨站请求中发送会话cookie。默认为 ‘lax’。
https_only - 应设置安全标志（只能与HTTPS一起使用）。默认为False。

CORSMiddleware：提供跨域资源共享支持

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

app = FastAPI()

origins = [
    "https://test.foxtells.com",
    "http://test.foxtells.com",
    "http://localhost",
    "http://localhost:8088",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def index():
    return "test middleware"

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

CORSMiddleware中间件有如下配置参数：
allow_origins - 允许跨域请求的域名列表，例如 [‘https://example.org’, ‘https://www.example.org’] 或者 [‘*’]。
allow_origin_regex - 允许跨域请求的域名正则表达式，例如 ‘https://.*.example.org’。
allow_methods - 允许跨域请求的HTTP方法列表，默认为[‘GET’]，[‘*’] 表示允许所有HTTP方法。
allow_headers - 跨域请求支持的HTTP头信息列表。[‘*’] 表示允许所有头信息。Accept, Accept-Language, Content-Language 和 Content-Type头信息默认全都支持。
allow_credentials - 表示在跨域请求时是否支持cookie，默认为False。
expose_headers - 表示对浏览器可见的返回结果头信息，默认为[]。
max_age - 浏览器缓存CORS返回结果的最大时长，默认为600(单位秒)。

HTTPSRedirectMiddleware：强制所有的请求使用http或者wss协议

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

app = FastAPI()

app.add_middleware(HTTPSRedirectMiddleware)


@app.get("/")
async def index():
    return "test middleware"
    
1
2
3
4
5
6
7
8
9
10
11
12

TrustedHostMiddleware：强制所有的请求来自于可信Host，为了防止HTTP Host的头部攻击

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

app = FastAPI()

app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["foxtells.com", "*.foxtells.com"]
)

@app.get("/")
async def index():
    return "test middleware"
1
2
3
4
5
6
7
8
9
10
11
12

GZipMiddleware：对内容进行GZip压缩

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

app = FastAPI()

app.add_middleware(GZipMiddleware, minimum_size=1000)


@app.get("/")
async def index():
    return "some very big size content"

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

CORSMiddleware中间件有如下配置参数：
minimum_size - 当返回结果大小小于指定值时不启用压缩。(单位为字节，默认值为500)

第三方提供的中间件

CSRFMiddleware：用于防御CSRF攻击的中间件。此中间件实现了双提交Cookie模式，即首先设置一个cookie，然后将其与HTTP头x-csrftoken进行比较。
安装中间件：

pip install starlette-csrf
1

基本使用：

from fastapi import FastAPI
from starlette_csrf import CSRFMiddleware

app = FastAPI()

app.add_middleware(CSRFMiddleware, secret="__CHANGE_ME__")

@app.get("/")
async def index():
    return "test middleware"

1
2
3
4
5
6
7
8
9
10
11

CSRFMiddleware中间件有如下配置参数：

secret (str) - 用于签名CSRF令牌值的密码。
required_urls(可选[List[re.Pattern]] - None) - 始终执行CSRF检查的URL正则表达式列表。
exempt_urls(可选[List[re.Pattern]] - None)- 跳过CSRF检查的URL正则表达式列表。sensitive_cookies (Set[str] - None) - 如果在请求中存在，应触发CSRF检查的cookie名称集合。如果此参数为None，即默认值，将始终执行CSRF。
safe_methods (Set[str] - {“GET”, “HEAD”, “OPTIONS”, “TRACE”}) - 不需要CSRF检查的HTTP方法。
cookie_name (str - csrftoken) - cookie的名称。
cookie_path (str - /) - cookie的路径。
cookie_domain (可选[str] - None) - cookie的域。如果您的前端和API位于不同的子域中，请确保使用您的根域设置此参数，以允许您的前端子域在JavaScript端读取cookie。
cookie_secure (bool - False) - 是否仅通过SSL请求将cookie发送到服务器。
cookie_samesite (str - lax) - cookie的Samesite策略。
header_name (str - x-csrftoken) - 设置CSRF令牌的header名称。

通过继承CSRFMiddleware，可以自定义错误Response：

from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette_csrf import CSRFMiddleware

class CustomResponseCSRFMiddleware(CSRFMiddleware):
    def _get_error_response(self, request: Request) -> Response:
        return JSONResponse(
            content={"code": "CSRF_ERROR"}, status_code=403
        )
1
2
3
4
5
6
7
8
9