Tartiflette + FastApi 身份验证

Question

我正在使用 tartiflette-asgi 使用 FastApi 构建一个 tartiflette 应用程序，但我找不到使常规 FastApi 身份验证或依赖注入工作的方法。

问题在于 tartiflette 应用程序是如何构建和安装的。做的时候

app = FastApi()

gql_app = TartifletteApp(..)
app.mount("/graphql", gql_app)

我无法指定依赖项来执行我的标头验证。我试过使用 FastApi include_router，但它根本不适用于 TartifletteApp。我也尝试过像

这样的小技巧

gql_app = TartifletteApp(..)

app.include_router(
    gql_app.router,
    prefix="/graphql",
    # dependencies=[Depends(get_current_user)],   # here I would add a token and get a user
)

我得到了错误

File "/usr/local/lib/python3.6/site-packages/uvicorn/protocols/http/h11_impl.py", line 389, in run_asgi
     result = await app(self.scope, self.receive, self.send)
   File "/usr/local/lib/python3.6/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
     return await self.app(scope, receive, send)
   File "/usr/local/lib/python3.6/site-packages/fastapi/applications.py", line 181, in __call__
     await super().__call__(scope, receive, send)  # pragma: no cover
   File "/usr/local/lib/python3.6/site-packages/starlette/applications.py", line 111, in __call__
     await self.middleware_stack(scope, receive, send)
   File "/usr/local/lib/python3.6/site-packages/starlette/middleware/errors.py", line 181, in __call__
     raise exc from None
   File "/usr/local/lib/python3.6/site-packages/starlette/middleware/errors.py", line 159, in __call__
     await self.app(scope, receive, _send)
   File "/usr/local/lib/python3.6/site-packages/starlette/exceptions.py", line 82, in __call__
     raise exc from None
   File "/usr/local/lib/python3.6/site-packages/starlette/exceptions.py", line 71, in __call__
     await self.app(scope, receive, sender)
   File "/usr/local/lib/python3.6/site-packages/starlette/routing.py", line 566, in __call__
     await route.handle(scope, receive, send)
   File "/usr/local/lib/python3.6/site-packages/starlette/routing.py", line 227, in handle
     await self.app(scope, receive, send)
   File "/usr/local/lib/python3.6/site-packages/tartiflette_asgi/_endpoints.py", line 84, in dispatch
     graphiql = get_graphql_config(request).graphiql
   File "/usr/local/lib/python3.6/site-packages/tartiflette_asgi/_middleware.py", line 18, in get_graphql_config
     config = conn["graphql"]
   File "/usr/local/lib/python3.6/site-packages/starlette/requests.py", line 68, in __getitem__
     return self.scope[key]
 KeyError: 'graphql'

我可以将标头验证实现为 graphql 中间件，但我希望我可以在 FastApi 级别执行此操作，以便它适用于每个端点。

关于如何解决这个问题的任何建议？

Answer 1

首先创建一个基本身份验证，然后将其添加到依赖项而不是获取当前用户，这将为您的端点添加一个基本身份验证

from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

app.include_router(
gql_app.router,
prefix="/graphql",
dependencies=[Depends(security)], 
)

Answer 2

我已经设法在没有 tartiflette-asgi 的情况下解决了这个问题。解决办法贴here

看起来像：

import os
import json
import typing

from starlette.background import BackgroundTasks
from starlette.datastructures import QueryParams
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response

from tartiflette import Engine


class GraphQLApp:

    def __init__(self, app, modules, schema_path=None, error_coercer=None):
        self.engine = Engine(
            sdl=schema_path or os.path.join(os.path.dirname(__file__), "schema"),
            modules=modules,
            error_coercer=error_coercer,
        )

        app.on_event("startup")(self._cook)

    async def _cook(self):
        await self.engine.cook()

    def _build_context(self, **kwargs):
        return kwargs or {}  # add custom logic when needed here

    async def _get_response(self, request: Request, data: QueryParams, context: dict) -> Response:
        try:
            query = data["query"]
        except KeyError:
            return PlainTextResponse("No GraphQL query found in the request", 400)

        def _format_error(error: typing.Any) -> dict:
            import ast

            try:
                return ast.literal_eval(str(error))
            except ValueError:
                return {"message": "Internal Server Error"}

        background = BackgroundTasks()
        context = {"req": request, "background": background, **self._build_context(**context)}

        result: dict = await self.engine.execute(
            query,
            context=context,
            variables=data.get("variables"),
            operation_name=data.get("operationName"),
        )

        content = {"data": result["data"]}
        has_errors = "errors" in result
        if has_errors:
            content["errors"] = [_format_error(error) for error in result["errors"]]

        status = 400 if has_errors else 200
        return JSONResponse(content=content, status_code=status, background=background)

    async def process_request(self, request: Request, context: dict = None) -> Response:
        content_type = request.headers.get("Content-Type", "")
        if "application/json" in content_type:
            try:
                data = await request.json()
            except json.JSONDecodeError:
                return JSONResponse({"error": "Invalid JSON."}, 400)

        elif "application/graphql" in content_type:
            body = await request.body()
            data = {"query": body.decode()}

        elif "query" in request.query_params:
            data = request.query_params

        else:
            return PlainTextResponse("Unsupported Media Type", 415)

        return await self._get_response(request, data=data, context=context or {})

这样我就可以了

app = FastApi()
gql_app = GraphQLApp(app)

@app.post("/graphql")
async def graphql_ninja(request: Request):
    return await gql_app.process_request(request)