5

我在使用 FastAPI 构建的 GraphQL 服务器中实现身份验证时遇到问题。以前,我们使用的是 REST,但现在我们正在切换到 GraphQL,我想知道如何实现它。之前,我们有不同的路由器,并且使用 FastAPI 可以很容易地使用依赖关系检查基于路由的身份验证,如下所示。我们在授权标头中发送一个令牌,我们在后端解码它并取回 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。最好的方法是什么?

4

1 回答 1

7

在他们使用的 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)])
于 2020-08-05T12:13:36.113 回答