搭建通用FastAPI网关（通用端点）

你的目标是实现一个支持全类型请求转发的API网关，当前实现的核心问题集中在头部处理、请求拼接逻辑、表单/文件处理以及异常处理这几个方面，以下是针对性的优化建议：

一、修复头部过滤与转发逻辑 #

1. 修正通配符头部过滤
   当前代码中proxy-*是精确匹配，无法过滤所有以proxy-开头的头部，需要改为前缀判断：

   headers_to_not_forward = ['host', 'connection', 'keep-alive', 'transfer-encoding', 'te', 'trailer', 'upgrade']
   forward_headers = {}
   for k, v in request.headers.items():
       lower_k = k.lower()
       if lower_k not in headers_to_not_forward and not lower_k.startswith('proxy-'):
           forward_headers[k] = v
   

2. 响应头部清理
   除了content-encoding，还需要移除transfer-encoding和connection这类连接相关的头部，避免客户端出现兼容性问题：

   @asynccontextmanager
   async def handle_api_response(response_coroutine):
       try:
           response = await response_coroutine
           # 移除httpx自动处理的编码和连接类头部
           for header in ["content-encoding", "transfer-encoding", "connection"]:
               if header in response.headers:
                   del response.headers[header]
           yield response
       except httpx.HTTPStatusError as e:
           # 目标服务返回4xx/5xx时，原样返回而非转成500
           yield e.response
       except Exception as e:
           raise HTTPException(status_code=500, detail=f"{e.__class__.__name__}: {e}")
   

二、优化请求路径与参数拼接 #

1. 安全拼接目标URL
   手动拼接路径容易出现双斜杠问题，改用urllib.parse.urljoin处理路径拼接，同时直接传递查询参数而非手动拼接字符串：

   from urllib.parse import urljoin
   
   target_url = urljoin(os.environ['TARGET_URL'], relative_path)
   # 直接传递查询参数，避免编码问题
   params = dict(request.query_params)
   

   在httpx请求中添加params=params参数。

2. 支持所有HTTP方法
   将路由的methods改为["*"]，覆盖所有HTTP请求方法：

   @app.api_route("/{relative_path:path}", methods=["*"])
   

三、修复表单与文件上传处理 #

1. 正确处理UploadFile流
   FastAPI的UploadFile对象在读取后指针会移动到末尾，需要重置指针后再传给httpx，同时确保文件被正确关闭：

   elif 'multipart/form-data' in content_type:
       form_data = await request.form()
       files = {}
       form = {}
       for k, v in form_data.items():
           if isinstance(v, UploadFile):
               # 重置文件指针到开头
               await v.seek(0)
               files[k] = (v.filename, v.file, v.content_type)
           else:
               form[k] = v
       # 请求结束后关闭文件
       try:
           response_coroutine = client.request(
               method=request.method,
               url=target_url,
               headers=forward_headers,
               data=form, 
               files=files,
               params=params
           )
           # ...后续处理
       finally:
           for v in files.values():
               v[1].close()
   

四、支持流式请求与响应 #

对于大文件上传/下载，避免一次性读取全部内容到内存，改用流式处理：

# 读取请求流
async def stream_request_body(request: Request):
    async for chunk in request.stream():
        yield chunk

# 在非表单请求中使用流式内容
else:
    response_coroutine = client.request(
        method=request.method,
        url=target_url,
        headers=forward_headers,
        content=stream_request_body(request),
        params=params,
        follow_redirects=False  # 不自动跟随重定向，原样返回给客户端
    )

五、完整优化后的代码 #

from fastapi import FastAPI, Request, Response, Path, UploadFile, HTTPException
import httpx
import os
from contextlib import asynccontextmanager
from urllib.parse import urljoin

app = FastAPI()

@app.api_route("/{relative_path:path}", methods=["*"])
async def apim_anonymous(request: Request, relative_path: str = Path(...)):
    # 过滤不需要转发的头部
    headers_to_not_forward = ['host', 'connection', 'keep-alive', 'transfer-encoding', 'te', 'trailer', 'upgrade']
    forward_headers = {}
    for k, v in request.headers.items():
        lower_k = k.lower()
        if lower_k not in headers_to_not_forward and not lower_k.startswith('proxy-'):
            forward_headers[k] = v

    content_type = request.headers.get('content-type', '')
    params = dict(request.query_params)
    target_url = urljoin(os.environ['TARGET_URL'], relative_path)

    async with httpx.AsyncClient() as client:
        if 'application/x-www-form-urlencoded' in content_type:
            form_data = await request.form()
            body = {key: value for key, value in form_data.items()}
            response_coroutine = client.request(
                method=request.method,
                url=target_url,
                headers=forward_headers,
                data=body,
                params=params,
                follow_redirects=False
            )
        elif 'multipart/form-data' in content_type:
            form_data = await request.form()
            files = {}
            form = {}
            for k, v in form_data.items():
                if isinstance(v, UploadFile):
                    await v.seek(0)
                    files[k] = (v.filename, v.file, v.content_type)
                else:
                    form[k] = v
            try:
                response_coroutine = client.request(
                    method=request.method,
                    url=target_url,
                    headers=forward_headers,
                    data=form,
                    files=files,
                    params=params,
                    follow_redirects=False
                )
            finally:
                for f in files.values():
                    f[1].close()
        else:
            # 流式处理请求体
            async def stream_body():
                async for chunk in request.stream():
                    yield chunk

            response_coroutine = client.request(
                method=request.method,
                url=target_url,
                headers=forward_headers,
                content=stream_body(),
                params=params,
                follow_redirects=False
            )

        async with handle_api_response(response_coroutine) as response:
            # 流式返回响应内容
            return Response(
                content=response.aiter_bytes(),
                status_code=response.status_code,
                headers=dict(response.headers),
                media_type=response.headers.get('content-type')
            )

@asynccontextmanager
async def handle_api_response(response_coroutine):
    try:
        response = await response_coroutine
        # 移除不需要转发的响应头部
        cleanup_headers = ["content-encoding", "transfer-encoding", "connection", "keep-alive"]
        for header in cleanup_headers:
            if header in response.headers:
                del response.headers[header]
        yield response
    except httpx.HTTPStatusError as e:
        # 原样返回目标服务的错误响应
        for header in ["content-encoding", "transfer-encoding", "connection", "keep-alive"]:
            if header in e.response.headers:
                del e.response.headers[header]
        yield e.response
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"{e.__class__.__name__}: {e}")

关键优化点总结 #

• 修复了头部过滤的通配符逻辑，确保所有proxy-前缀头部被过滤
• 使用urljoin安全拼接路径，避免双斜杠问题
• 支持流式请求/响应，适配大文件传输场景
• 正确处理UploadFile的文件指针与关闭逻辑
• 保留目标服务的原始错误响应，而非统一返回500
• 禁用httpx自动跟随重定向，让客户端处理重定向逻辑

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