FastAPI 中间件实战：构建强大的 API 应用 – wiki基地

作者： /

FastAPI 中间件实战：构建强大的 API 应用

FastAPI 作为一款现代、高性能的 Python Web 框架，凭借其简洁的语法、强大的依赖注入系统、以及自动化的 API 文档生成能力，迅速获得了开发者的青睐。除了这些特性，FastAPI 还提供了强大的中间件支持，允许开发者在请求到达路由处理函数之前或之后执行自定义逻辑，从而实现各种通用的功能，例如身份验证、日志记录、请求转换、性能监控等等。本文将深入探讨 FastAPI 中间件的概念、类型、以及实战应用，帮助你构建更强大、更健壮的 API 应用。

一、理解 FastAPI 中间件

中间件（Middleware）本质上是一个处于请求和响应之间的函数或类，它拦截每一个进入应用的请求，并可以在处理请求之前、之后，甚至完全短路请求流程。它可以修改请求对象、执行预处理逻辑、生成响应，或者将请求传递给下一个中间件或路由处理函数。

在 FastAPI 中，中间件的注册和使用非常简单。你可以将中间件函数或类注册到 FastAPI 应用实例上，然后 FastAPI 会按照注册的顺序依次调用这些中间件。

中间件的优势：

  • 代码复用： 中间件可以封装通用的逻辑，并在多个路由之间共享，避免代码重复。
  • 模块化： 中间件将应用程序的不同关注点分离，例如身份验证、日志记录等，提高了代码的可维护性和可测试性。
  • 灵活性： 中间件允许你灵活地修改请求和响应，或者完全短路请求流程，满足不同的业务需求。
  • 性能优化： 中间件可以执行缓存、压缩等操作，优化应用程序的性能。
  • 安全增强： 中间件可以执行身份验证、授权等操作，增强应用程序的安全性。

二、FastAPI 中间件的类型

FastAPI 支持两种类型的中间件：

  1. 函数型中间件： 使用 async def 定义的异步函数，接收 request 和 call_next 两个参数。
  2. 类中间件： 定义包含 dispatch 方法的类，dispatch 方法接收 request 和 call_next 两个参数。

1. 函数型中间件：

函数型中间件是最常用的中间件类型。它使用 async def 定义异步函数，接收 request 和 call_next 两个参数。

  • request: Request 对象，包含请求的所有信息，例如请求头、查询参数、请求体等。
  • call_next: 一个可调用对象（通常是另一个中间件或路由处理函数），用于将请求传递给下一个处理者。

示例：日志记录中间件（函数型）

“`python
from fastapi import FastAPI, Request
import time

app = FastAPI()

async def log_middleware(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() – start_time
print(f”Request: {request.url}, Method: {request.method}, Process Time: {process_time:.4f}s”)
return response

app.middleware(“http”, log_middleware)

@app.get(“/”)
async def read_root():
return {“message”: “Hello World”}
“`

在这个例子中，log_middleware 函数拦截了每一个 HTTP 请求。它记录了请求的 URL、请求方法以及处理请求所需的时间。 call_next(request) 负责将请求传递给下一个中间件或路由处理函数，并返回响应。

2. 类中间件：

类中间件使用包含 dispatch 方法的类来定义。dispatch 方法接收 request 和 call_next 两个参数，与函数型中间件的参数含义相同。 类中间件允许你拥有状态，并且可以在中间件中使用依赖注入。

示例：认证中间件（类中间件）

“`python
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

class AuthenticationMiddleware:
def init(self, app):
self.app = app

async def __call__(self, request: Request, call_next):
    if request.url.path.startswith("/protected"):
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse(content={"message": "Unauthorized"}, status_code=401)

        token = auth_header.split(" ")[1]
        if token != "valid_token":
            return JSONResponse(content={"message": "Unauthorized"}, status_code=401)

    response = await call_next(request)
    return response

app.add_middleware(AuthenticationMiddleware)

@app.get(“/”)
async def read_root():
return {“message”: “Hello World”}

@app.get(“/protected”)
async def read_protected():
return {“message”: “This is a protected resource”}
“`

在这个例子中，AuthenticationMiddleware 类拦截了所有以 /protected 开头的请求。它检查请求头中是否包含 Authorization 头，并且验证 token 是否有效。如果 token 无效，它会返回一个 401 Unauthorized 响应。

三、中间件的注册和排序

FastAPI 提供了两种注册中间件的方式：

  • app.middleware("http", middleware_function): 这种方式直接将函数型中间件注册到应用实例上。"http" 表示该中间件处理所有 HTTP 请求。
  • app.add_middleware(MiddlewareClass): 这种方式将类中间件添加到应用实例上。

中间件的排序：

中间件的执行顺序非常重要。 FastAPI 按照注册的顺序依次调用这些中间件。这意味着，你先注册的中间件会先执行。 通常，认证中间件应该在日志记录中间件之前执行，以便在记录请求之前验证用户的身份。

四、实战：构建强大的 API 应用

下面我们将通过几个具体的例子，展示如何使用中间件构建强大的 API 应用。

1. 认证授权：

认证授权是 API 应用中非常重要的组成部分。使用中间件可以很方便地实现认证授权功能。

示例：使用 JWT 进行认证授权

“`python
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import jwt
from datetime import datetime, timedelta

app = FastAPI()
SECRET_KEY = “your_secret_key”
ALGORITHM = “HS256”

async def authenticate_middleware(request: Request, call_next):
if request.url.path.startswith(“/protected”):
auth_header = request.headers.get(“Authorization”)
if not auth_header or not auth_header.startswith(“Bearer “):
return JSONResponse(content={“message”: “Unauthorized”}, status_code=401)

    token = auth_header.split(" ")[1]
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        request.state.user_id = payload.get("sub") # Store user id in request state
    except jwt.ExpiredSignatureError:
        return JSONResponse(content={"message": "Token expired"}, status_code=401)
    except jwt.InvalidTokenError:
        return JSONResponse(content={"message": "Invalid token"}, status_code=401)

response = await call_next(request)
return response

app.middleware(“http”, authenticate_middleware)

@app.post(“/login”)
async def login(username: str, password: str):
# In a real application, you would authenticate against a database
if username == “user” and password == “password”:
payload = {
“sub”: “user123”,
“exp”: datetime.utcnow() + timedelta(minutes=30)
}
token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
return {“access_token”: token, “token_type”: “bearer”}
else:
raise HTTPException(status_code=401, detail=”Invalid credentials”)

@app.get(“/protected”)
async def read_protected(request: Request):
user_id = request.state.user_id
return {“message”: f”This is a protected resource, user ID: {user_id}”}
“`

在这个例子中：

  • /login 路由处理用户的登录请求，生成 JWT token 并返回。
  • authenticate_middleware 拦截所有以 /protected 开头的请求，验证 JWT token 的有效性，并将用户 ID 存储在 request.state 中，以便后续路由处理函数可以使用。
  • /protected 路由处理受保护的资源请求，从 request.state 中获取用户 ID。

2. 请求转换：

中间件可以用于修改请求对象，例如将请求体转换为不同的格式，或者添加额外的请求头。

示例：将请求体转换为 JSON 格式

“`python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import json

app = FastAPI()

async def json_middleware(request: Request, call_next):
if request.method == “POST” or request.method == “PUT”:
try:
body = await request.body()
request.state.body = json.loads(body.decode(“utf-8”))
except json.JSONDecodeError:
return JSONResponse(content={“message”: “Invalid JSON body”}, status_code=400)

response = await call_next(request)
return response

app.middleware(“http”, json_middleware)

@app.post(“/items”)
async def create_item(request: Request):
item = request.state.body
# Process the item
return {“message”: “Item created”, “item”: item}
“`

在这个例子中，json_middleware 拦截所有 POST 和 PUT 请求，尝试将请求体解析为 JSON 格式，并将解析后的 JSON 对象存储在 request.state.body 中。 如果解析失败，则返回 400 Bad Request 响应。

3. 性能监控：

中间件可以用于监控 API 的性能，例如记录请求的处理时间、CPU 使用率、内存使用率等。

示例：记录请求处理时间

“`python
from fastapi import FastAPI, Request
import time

app = FastAPI()

async def performance_middleware(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() – start_time
# Log the process time to a monitoring system
print(f”Request: {request.url}, Process Time: {process_time:.4f}s”)
return response

app.middleware(“http”, performance_middleware)

@app.get(“/”)
async def read_root():
return {“message”: “Hello World”}
“`

在这个例子中，performance_middleware 记录了每个请求的处理时间，并将处理时间打印到控制台。你可以将这些数据发送到监控系统，以便实时监控 API 的性能。

4. CORS (跨域资源共享)

CORS 是一个安全机制，用于限制跨域请求。 FastAPI 提供了内置的 CORS 中间件，可以方便地配置 CORS 策略。

“`python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
“http://localhost”,
“http://localhost:8080”,
“https://your-frontend-domain.com”,
]

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

@app.get(“/”)
async def read_root():
return {“message”: “Hello World”}
“`

在这个例子中，CORSMiddleware 允许来自 origins 列表中域名的跨域请求。 allow_origins 指定允许的域名， allow_credentials 允许跨域请求携带 cookie， allow_methods 指定允许的 HTTP 方法， allow_headers 指定允许的请求头。

五、总结

FastAPI 中间件是一种强大的工具，可以帮助你构建更强大、更健壮的 API 应用。 通过使用中间件，你可以封装通用的逻辑，分离应用程序的不同关注点，提高代码的可维护性和可测试性，优化应用程序的性能，并增强应用程序的安全性。 本文介绍了 FastAPI 中间件的概念、类型、注册和排序方式，并通过几个具体的例子展示了如何使用中间件实现认证授权、请求转换、性能监控和 CORS 等功能。 希望本文能够帮助你更好地理解和使用 FastAPI 中间件，构建出高质量的 API 应用。

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部