Source code for swh.graphql.server

# Copyright (C) 2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

import os
from typing import Any, Dict, Optional

from aiocache import Cache
from ariadne.asgi import GraphQL
from starlette.applications import Starlette
from starlette.authentication import (
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.middleware.cors import CORSMiddleware
from starlette.routing import Mount, Route
from starlette.staticfiles import StaticFiles

from swh.auth.starlette.backends import BearerTokenAuthBackend
from swh.core import config
from import get_search as get_swh_search
from import SearchInterface
from import get_storage as get_swh_storage
from import StorageInterface

from .client.view import explorer_page

graphql_cfg: Dict[str, Any] = {}
storage: Optional[StorageInterface] = None
search: Optional[SearchInterface] = None

[docs] def get_storage() -> StorageInterface: global storage if not storage: storage = get_swh_storage(**graphql_cfg["storage"]) return storage
[docs] def get_config() -> Dict[str, Any]: global graphql_cfg if not graphql_cfg: config_path = os.environ.get("SWH_CONFIG_FILENAME") graphql_cfg = load_and_check_config(config_path) return graphql_cfg
[docs] class AnonymousAuthBackend(AuthenticationBackend):
[docs] async def authenticate(self, conn): return AuthCredentials(["anonymous"]), UnauthenticatedUser()
[docs] def load_and_check_config(config_path: Optional[str]) -> Dict[str, Any]: """Check the minimal configuration is set to run the api or raise an error explanation. Args: config_path: Path to the configuration file to load Raises: Error if the setup is not as expected Returns: configuration as a dict """ if not config_path: raise EnvironmentError("Configuration file must be defined") if not os.path.exists(config_path): raise FileNotFoundError(f"Configuration file {config_path} does not exist") cfg = if "storage" not in cfg: raise KeyError("Missing 'storage' configuration") return cfg
[docs] def make_app_from_configfile(): """Loading the configuration from a configuration file. SWH_CONFIG_FILENAME environment variable defines the configuration path to load. """ from .app import schema, validation_rules from .errors import format_error, on_auth_error from .middlewares.logger import LogMiddleware graphql_cfg = get_config() ariadne_app = GraphQL( schema, debug=graphql_cfg["debug"], introspection=graphql_cfg["introspection"], validation_rules=validation_rules, error_formatter=format_error, ) if "auth" in graphql_cfg: auth_backend = BearerTokenAuthBackend( server_url=graphql_cfg["auth"]["server"], realm_name=graphql_cfg["auth"]["realm"], client_id=graphql_cfg["auth"]["client"], # FIXME, improve this with response cache implementation cache=Cache.from_url(url=graphql_cfg["auth"]["cache"]["url"]), ) else: auth_backend = AnonymousAuthBackend() middleware = [ Middleware( CORSMiddleware, # FIXME, restrict origins after deploying the JS client allow_origins=["*"], allow_methods=("GET", "POST", "OPTIONS"), allow_headers=["*"], ), Middleware( AuthenticationMiddleware, backend=auth_backend, on_error=on_auth_error, ), Middleware(LogMiddleware), ] # Mount under a starlette application application = Starlette( routes=[ Route("/", ariadne_app, methods=["POST"], name="graphql_api"), Route("/", explorer_page, methods=["GET"]), Mount( "/static", app=StaticFiles(packages=[("swh.graphql.client", "static")]), name="static", ), ], middleware=middleware, ) return application