FastAPI多租户场景下动态修改路径参数customer_id的优化方案咨询

方案1：全局依赖统一处理ID偏移 #

FastAPI的依赖系统可以实现全局或路由组级别的参数预处理，无需逐个修改路由函数。

实现步骤： #

1. 定义依赖函数，识别租户并调整customer_id：

from fastapi import Depends, Request

async def get_adjusted_customer_id(request: Request) -> int:
    # 替换为你的租户识别逻辑，比如从请求头、API密钥获取租户标识
    is_new_tenant = await check_tenant_identity(request)
    
    original_customer_id = int(request.path_params.get("customer_id", 0))
    return original_customer_id + 5000 if is_new_tenant else original_customer_id

2. 在路由中使用该依赖替代原始参数：

class ReportsRouter:
    @router.post(
        "/{customer_id}/report", response_model=ReportResponse,
    )
    async def report(
        adjusted_customer_id: int = Depends(get_adjusted_customer_id),
        request: ReportRequest
    ):
        return await ReportGenerator(adjusted_customer_id, request)

3. 若要全局应用到所有路由，可在创建FastAPI实例时注册：

from fastapi import FastAPI

app = FastAPI(dependencies=[Depends(get_adjusted_customer_id)])
# 后续路由函数直接使用adjusted_customer_id参数即可

方案2：自定义路径参数类型 #

通过自定义Pydantic类型，让FastAPI在解析路径参数时自动完成ID偏移调整：

from fastapi import Path, Request
from pydantic import GetCoreSchemaHandler
from pydantic_core import core_schema
from typing import Annotated, Any

class AdjustedCustomerId(int):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        base_schema = handler(int)
        return core_schema.no_info_wrap_validator_function(
            cls._adjust_id,
            base_schema,
        )
    
    @classmethod
    def _adjust_id(cls, value: int) -> int:
        # 提前用轻量中间件将租户标识存入request.state，避免修改路径参数
        request = Request(None)
        is_new_tenant = request.state.is_new_tenant
        
        return value + 5000 if is_new_tenant else value

# 在路由中使用自定义类型
class ReportsRouter:
    @router.post(
        "/{customer_id}/report", response_model=ReportResponse,
    )
    async def report(
        customer_id: Annotated[AdjustedCustomerId, Path()],
        request: ReportRequest
    ):
        # 此处customer_id已为调整后的数值
        return await ReportGenerator(customer_id, request)

方案3：自定义路由装饰器 #

用装饰器包裹路由处理函数，自动调整customer_id，无需修改函数参数结构：

from functools import wraps

def adjust_customer_id(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        original_id = kwargs.get("customer_id")
        if original_id is not None:
            # 从路由实例或请求对象中获取租户标识
            is_new_tenant = await check_tenant_identity(args[0].request)
            kwargs["customer_id"] = original_id + 5000 if is_new_tenant else original_id
        return await func(*args, **kwargs)
    return wrapper

# 在路由上应用装饰器
class ReportsRouter:
    @router.post(
        "/{customer_id}/report", response_model=ReportResponse,
    )
    @adjust_customer_id
    async def report(customer_id: int, request: ReportRequest):
        return await ReportGenerator(customer_id, request)

关键提示： #

• 租户识别逻辑是核心，需根据业务场景实现（如从请求头X-Tenant-ID、域名、API密钥判断），建议封装为独立函数。
• 方案1的全局依赖是最简洁的实现方式，无需大量修改现有路由结构。

内容的提问来源于stack exchange，提问作者Sidharth Samant