Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/resolvers/base_node.py
| # Copyright (C) 2022 The Software Heritage developers | # Copyright (C) 2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from abc import ABC | |||||
| from collections import namedtuple | from collections import namedtuple | ||||
| from typing import Any, Optional, Union | |||||
| from graphql.type import GraphQLResolveInfo | |||||
| from swh.graphql import resolvers as rs | |||||
| from swh.graphql.backends.archive import Archive | from swh.graphql.backends.archive import Archive | ||||
| from swh.graphql.errors import ObjectNotFoundError | from swh.graphql.errors import ObjectNotFoundError | ||||
| class BaseNode(ABC): | class BaseNode: | ||||
| """ | """ | ||||
| Base resolver for all the nodes | Base resolver for all the nodes | ||||
| """ | """ | ||||
| def __init__(self, obj, info, node_data=None, **kwargs): | def __init__(self, obj, info, node_data: Optional[Any] = None, **kwargs): | ||||
| self.obj = obj | self.obj: Optional[Union[BaseNode, rs.base_connection.BaseConnection]] = obj | ||||
| self.info = info | self.info: GraphQLResolveInfo = info | ||||
| self.kwargs = kwargs | self.kwargs = kwargs | ||||
| # initialize commonly used vars | |||||
| self.archive = Archive() | self.archive = Archive() | ||||
| self._node = self._get_node(node_data) | self._node: Optional[Any] = self._get_node(node_data) | ||||
| # handle the errors, if any, after _node is set | # handle the errors, if any, after _node is set | ||||
| self._handle_node_errors() | self._handle_node_errors() | ||||
| def _get_node(self, node_data): | def _get_node(self, node_data: Optional[Any]) -> Optional[Any]: | ||||
| """ | """ | ||||
| Get the node object from the given data | Get the node object from the given data | ||||
| if the data (node_data) is none make a function call | if the data (node_data) is none make a function call | ||||
| to get data from backend | to get data from backend | ||||
| """ | """ | ||||
| if node_data is None: | if node_data is None: | ||||
| node_data = self._get_node_data() | node_data = self._get_node_data() | ||||
| return self._get_node_from_data(node_data) | return self._get_node_from_data(node_data) | ||||
| def _get_node_from_data(self, node_data): | def _get_node_from_data(self, node_data: Any) -> Optional[Any]: | ||||
| """ | """ | ||||
| Get the object from node_data | Get the object from node_data | ||||
| In case of a dict, convert it to an object | In case of a dict, convert it to an object | ||||
| Override to support different data structures | Override to support different data structures | ||||
| """ | """ | ||||
| if type(node_data) is dict: | if type(node_data) is dict: | ||||
| return namedtuple("NodeObj", node_data.keys())(*node_data.values()) | return namedtuple("NodeObj", node_data.keys())(*node_data.values()) | ||||
| return node_data | return node_data | ||||
| def _handle_node_errors(self): | def _handle_node_errors(self) -> None: | ||||
| """ | """ | ||||
| Handle any error related to node data | Handle any error related to node data | ||||
| raise an error in case the object returned is None | raise an error in case the object returned is None | ||||
| override for specific behaviour | override for specific behaviour | ||||
| """ | """ | ||||
| if self._node is None: | if self._node is None: | ||||
| raise ObjectNotFoundError("Requested object is not available") | raise ObjectNotFoundError("Requested object is not available") | ||||
| def _get_node_data(self): | def _get_node_data(self) -> Optional[Any]: | ||||
| """ | """ | ||||
| Override for desired behaviour | Override for desired behaviour | ||||
| This will be called only when node_data is None | This will be called only when node_data is None | ||||
| """ | """ | ||||
| # FIXME, make this call async (not for v1) | # FIXME, make this call async (not for v1) | ||||
| return None | return None | ||||
| def __getattr__(self, name): | def __getattr__(self, name: str) -> Any: | ||||
| """ | """ | ||||
| Any property defined in the sub-class will get precedence over | Any property defined in the sub-class will get precedence over | ||||
| the _node attributes | the _node attributes | ||||
| """ | """ | ||||
| return getattr(self._node, name) | return getattr(self._node, name) | ||||
| def is_type_of(self): | def is_type_of(self) -> str: | ||||
| return self.__class__.__name__ | return self.__class__.__name__ | ||||
| class BaseSWHNode(BaseNode): | class BaseSWHNode(BaseNode): | ||||
| """ | """ | ||||
| Base resolver for all the nodes with a SWHID field | Base resolver for all the nodes with a SWHID field | ||||
| """ | """ | ||||
| @property | @property | ||||
| def swhid(self): | def swhid(self): | ||||
| return self._node.swhid() | return self._node.swhid() | ||||