如何在FastAPI中实现服务器间请求转发与重复处理？

直接给你一个可行的实现方案，用装饰器就能完成请求同步转发的需求，完全适配FastAPI的异步场景：

首先安装依赖（需要用异步HTTP客户端发送转发请求）：

pip install aiohttp

装饰器代码实现 #

import aiohttp
import asyncio
from functools import wraps
from fastapi import Request, Response

def duplicate_request(target_server: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            # 先读取请求体（FastAPI的request是流式对象，只能读取一次）
            body = await request.body()
            
            # 并行执行原业务逻辑和转发请求（同时处理，符合你的需求）
            async def forward_req():
                url = f"{target_server}{request.url.path}?{request.url.query}"
                headers = dict(request.headers)
                headers.pop("Host", None)  # 移除原Host头，避免目标服务器识别异常
                
                try:
                    async with aiohttp.ClientSession() as session:
                        async with session.request(
                            method=request.method,
                            url=url,
                            headers=headers,
                            data=body
                        ) as resp:
                            # 可以在这里处理转发响应，比如记录日志
                            # print(f"转发请求响应状态：{resp.status}")
                            await resp.text()
                except Exception as e:
                    # 转发失败不影响原接口，只记录错误
                    print(f"转发请求到{target_server}失败：{str(e)}")

            # 同时执行原函数和转发请求
            main_response, _ = await asyncio.gather(
                func(request, *args, **kwargs),
                forward_req()
            )
            
            return main_response
        return wrapper
    return decorator

在路由中使用装饰器 #

from fastapi import APIRouter, Depends
from pydantic import BaseModel

router = APIRouter()

# 你的请求模型
class RegisterBookRequest(BaseModel):
    title: str
    author: str

# 模拟数据库依赖
def get_db():
    return "本地数据库连接"

@router.post("/register_book")
@duplicate_request("http://<第二台服务器IP>:3000")  # 替换成实际的服务器地址
async def register_book(
    book_data: RegisterBookRequest,
    db = Depends(get_db),
    request: Request = None  # 需要注入Request对象来获取请求信息
) -> Response:
    # 原有的业务逻辑，比如保存书籍到本地数据库
    print(f"本地处理：保存书籍《{book_data.title}》到数据库")
    return Response(content="书籍注册成功", status_code=200)

关键细节说明 #

• 用asyncio.gather实现并行执行，原接口处理和转发请求同时进行，不会因为转发拖慢原接口响应
• 提前读取请求体，避免流式对象被原函数消费后无法转发
• 转发失败不会影响原接口的正常响应，仅记录错误
• 自动携带原请求的路径、查询参数、请求头和请求体，保证转发的请求和原请求完全一致

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