如何在FastAPI与Strawberry构建的GraphQL API中实现特定状态码错误处理

GraphQL 默认返回 200 状态码，但要匹配业务场景返回对应 HTTP 状态码，需结合 Strawberry 的扩展机制和 FastAPI 的异常体系实现，以下是针对你需求的分步实现：

1. 定义业务异常类 #

先创建对应各错误场景的自定义异常，绑定目标 HTTP 状态码：

from fastapi import HTTPException

class ResourceNotFoundError(HTTPException):
    def __init__(self, detail: str = "资源未找到"):
        super().__init__(status_code=404, detail=detail)

class InputValidationError(HTTPException):
    def __init__(self, detail: str = "输入数据验证失败"):
        super().__init__(status_code=422, detail=detail)

class UnauthorizedError(HTTPException):
    def __init__(self, detail: str = "未授权访问"):
        super().__init__(status_code=401, detail=detail)

class ForbiddenError(HTTPException):
    def __init__(self, detail: str = "权限不足，禁止访问"):
        super().__init__(status_code=403, detail=detail)

2. 实现 Strawberry 错误处理扩展 #

通过 Strawberry 的 Extension 捕获执行阶段的异常，将自定义异常映射到对应 HTTP 状态码，同时兜底处理未捕获的 500 错误：

from strawberry.extensions import Extension
from strawberry.types import ExecutionContext
from starlette.requests import Request
from starlette.responses import JSONResponse

class HTTPStatusExtension(Extension):
    def on_request_start(self):
        self.request: Request = self.execution_context.context["request"]
    
    def on_execution_error(self, errors, execution_context: ExecutionContext):
        # 提取首个触发的异常（GraphQL 会收集所有错误，这里取核心错误）
        exception = errors[0].original_error
        if isinstance(exception, HTTPException):
            # 覆盖默认响应，设置对应状态码
            self.request.scope["response"] = JSONResponse(
                content={"errors": [{"message": exception.detail}]},
                status_code=exception.status_code
            )
        else:
            # 未捕获的异常统一返回 500
            self.request.scope["response"] = JSONResponse(
                content={"errors": [{"message": "服务器意外错误"}]},
                status_code=500
            )

3. 注册扩展并编写业务逻辑 #

创建 Schema 时注入自定义扩展，在业务字段中抛出对应异常：

import strawberry
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class Query:
    @strawberry.field
    def get_user(self, user_id: int) -> str:
        if user_id == 0:
            raise ResourceNotFoundError(detail=f"用户ID {user_id} 不存在")
        if user_id < 0:
            raise InputValidationError(detail=f"用户ID {user_id} 不能为负数")
        # 模拟认证检查
        if user_id == 100:
            raise UnauthorizedError()
        # 模拟权限检查
        if user_id == 200:
            raise ForbiddenError()
        return f"用户 {user_id}"

schema = strawberry.Schema(Query, extensions=[HTTPStatusExtension])
graphql_app = GraphQLRouter(schema)

4. 集成到 FastAPI 应用 #

将 GraphQL 路由挂载到 FastAPI 实例：

from fastapi import FastAPI

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

关键细节补充 #

• 输入验证增强：结合 Strawberry 的 @strawberry.input 和 Pydantic 验证，当输入不符合规则时主动抛出 InputValidationError，示例：

from strawberry import input
from pydantic import ValidationError, PositiveInt

@input
class UserInput:
    user_id: PositiveInt

@strawberry.type
class Query:
    @strawberry.field
    def get_user(self, input: UserInput) -> str:
        try:
            # 触发 Pydantic 验证（Strawberry 会自动执行，此处为显式示例）
            input.user_id
        except ValidationError as e:
            raise InputValidationError(detail=str(e))
        # 业务逻辑
        return f"用户 {input.user_id}"

• 认证权限前置：通过 FastAPI 的 Depends 实现全局/局部认证，未通过时抛出对应异常：

from fastapi import Depends

def get_current_user(token: str = Depends(oauth2_scheme)):
    # 认证逻辑
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise UnauthorizedError(detail="无效的认证令牌")
    except JWTError:
        raise UnauthorizedError()
    # 权限检查
    if not check_permission(username):
        raise ForbiddenError()
    return username

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