FastAPI集成Supabase Storage Auth时Depends引发JWT格式错误问题

问题场景与代码 #

配置代码 #

def get_user_token(request: Request, x_user_token: str = Header(None)):
    return x_user_token

def get_storage(token: str = Depends(get_user_token)):
    supabase.auth._headers.update({"Authorization": f"Bearer {token}"})
    return supabase.storage

class Service:
    def __init__(self, db=Depends(get_db), storage=Depends(get_storage)):
        self.db = db
        self.storage = storage

    def create_service(cls, token: str = None):
        db = get_db(token=token) if token else get_db()
        storage = get_storage(token=token) if token else get_storage()
        return cls(db=db, storage=storage)

class FileService(Service):
    def upload(self, file_name: str, file_bytes: bytes):
        self.storage.from_(config.bucket).upload(
            f"{file_name.__str__()}.csv",
            file_bytes,
        )

FastAPI接口代码 #

@router.post("/endpoint")
def create_test(
    name: str,
    file_service: FileService = Depends(FileService),
):
 
    file_service.upload(
        file_name=name, file_bytes=...
    )

错误信息 #

storage3.utils.StorageException: {'statusCode': 400, 'error': 'Unauthorized', 'message': 'jwt malformed'}

用户疑问 #

是否因为在FastAPI外部使用了Depends导致的错误？

问题原因 #

核心问题有三点：

1. Depends仅在FastAPI请求上下文生效：当你在Service.create_service里直接调用get_storage()（未传token）时，get_storage函数中token: str = Depends(get_user_token)的Depends不会被FastAPI解析，此时token的值是Depends对象本身，而非实际的用户token。拼接成Authorization头后就变成了Bearer <Depends object>，自然触发"jwt malformed"错误。
2. 全局客户端头信息污染：直接修改supabase.auth._headers是全局操作，并发请求时会导致不同请求的token互相覆盖，也可能引发授权异常。
3. create_service缺少类方法装饰器：原代码中create_service方法没有@classmethod，调用cls()会抛出语法错误，这是隐藏的潜在问题。

解决方案 #

1. 重构依赖函数，支持手动传参 #

调整get_storage，让它既可以通过FastAPI依赖注入获取token，也支持手动传入token，同时避免修改全局变量：

def get_storage(token: str = Depends(get_user_token)):
    # 复制原客户端的头信息，避免并发请求互相干扰
    storage_client = supabase.storage
    storage_client._headers = storage_client._headers.copy()
    storage_client._headers.update({"Authorization": f"Bearer {token}"})
    return storage_client

2. 修复create_service方法 #

添加@classmethod装饰器，并确保手动传token时直接传递有效值：

class Service:
    def __init__(self, db=Depends(get_db), storage=Depends(get_storage)):
        self.db = db
        self.storage = storage

    @classmethod
    def create_service(cls, token: str = None):
        if token:
            # 手动传入token，跳过依赖注入逻辑
            db = get_db(token=token)
            storage = get_storage(token=token)
        else:
            # 此分支仅在FastAPI请求上下文内使用，否则Depends无法解析
            db = get_db()
            storage = get_storage()
        return cls(db=db, storage=storage)

3. 接口无需修改 #

原FastAPI接口的FileService = Depends(FileService)会让FastAPI自动解析构造函数中的依赖，在请求上下文内正确获取x_user_token并传递给get_storage，可以正常工作。

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