使用 FastAPI 在基于 Python 的 GraphQL 服务器中进行身份验证

Question

我在使用 FastAPI 构建的 GraphQL 服务器中实现身份验证时遇到问题。以前，我们使用的是 REST，但现在我们正在切换到 GraphQL，我想知道如何实现它。以前，我们有不同的路由器，并且使用 FastAPI 可以很容易地使用 here 中的依赖关系检查基于路由的身份验证。我们在授权标头中发送一个令牌，我们在后端对其进行解码并取回 user_id，然后我们可以在不同的端点中使用它。

我想知道这里如何使用 GraphQL。我们使用Graphene，我查看了Starlettes Authentication Examples 以及设置GraphQl 的介绍

import binascii
from fastapi import FastAPI
from starlette.authentication import (
    AuthenticationBackend, AuthenticationError, SimpleUser, AuthCredentials
)
from starlette.graphql import GraphQLApp
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware

from schemas.root import my_schema


class BasicAuthBackend(AuthenticationBackend):
    async def authenticate(self, request):
        if "Authorization" not in request.headers:
            raise AuthenticationError('No auth credentials')

        auth = request.headers["Authorization"]
        try:
            id_token = auth.split('Bearer ')[1]
            decoded_token = auth.verify_id_token(id_token)

        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
            raise AuthenticationError('Invalid basic auth credentials')

        user_id = decoded_token['uid']
        return AuthCredentials(["authenticated"]), user_id


middleware = [
    Middleware(AuthenticationMiddleware, backend=BasicAuthBackend())
]

my_schema = Schema(
    query=RootQuery,
    mutation=RootMutation,
)

api = FastAPI(title=f"MyGraphQLServer", middleware=middleware)
api.add_route("/graphql", GraphQLApp(schema=my_schema))

例如，假设我现在只想验证突变请求而不是查询请求。此外，我想访问每个解析器中的 user_id。最好的方法是什么？

Answer 1

在 FastAPI 文档或 starlette 文档中，他们使用 add_route，这是在 Starlette 中添加路由而不声明特定操作的方式（如 .get()、.post() 等）。但是它有一些缺点，我们不能像在 FastAPI 中那样添加依赖，下面的例子

app.add_route(
"/graphql",
GraphQLApp(schema=graphene.Schema(query=Query), 
executor_class=AsyncioExecutor),
    dependencies=(Depends(SomeAuthorizationStuffHere)),
)

所以我们需要在 FastAPI 中做，我用 HTTPBasicAuth 创建了一个简单的应用程序，你可以用其他方法扩展它，你只需要包含router(s)

from fastapi import Query, Depends, Request, FastAPI, APIRouter
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import graphene
from graphene import Field, Schema, String, ObjectType
from starlette.graphql import GraphQLApp


router = APIRouter()
app = FastAPI()
security = HTTPBasic()


class Query(ObjectType):
    hello = Field(String, name=String())

    def resolve_hello(root, info, name):
        coverage = info.context["request"].state.some_dep
        return f"Hello {some_dep.some_method(name)}"


graphql_app = GraphQLApp(schema=Schema(query=Query))


@router.api_route("/gql", methods=["GET", "POST"])
async def graphql(request: Request):
    return await graphql_app.handle_graphql(request=request)


app.include_router(router, dependencies=[Depends(security)])