diff --git a/swh/graphql/resolvers/base_connection.py b/swh/graphql/resolvers/base_connection.py index f64fb25..0cce3df 100644 --- a/swh/graphql/resolvers/base_connection.py +++ b/swh/graphql/resolvers/base_connection.py @@ -1,119 +1,120 @@ -""" -""" from abc import ABC, abstractmethod from typing import Any from swh.graphql.utils import utils # from dataclasses import dataclass # @dataclass # class PageInfo: # nex_page_token: str # class Arguments: # """ # dataclass # """ # after # Elements that come after the specified cursor # first # Returns the first n elements class BaseConnection(ABC): + """ + Base class for all the connection resolvers + """ _node_class: Any = None - _page_size = 50 # deafult page size + _page_size = 50 # default page size def __init__(self, obj, info, **kwargs): self.obj = obj self.info = info self.kwargs = kwargs self._paged_data = None - self.pageInfo = self.page_info - self.totalCount = self.total_count + self.pageInfo = self.page_info # To match the name in schema + self.totalCount = self.total_count # To match the name in schema def __call__(self, *args, **kw): return self @property def edges(self): return self._get_edges() @property def nodes(self): """ Override if needed return a list of objects If a node class is set, return a list of its (Node) intances else a list of raw results """ if self._node_class is not None: return [ self._node_class(self.obj, self.info, node_data=result, **self.kwargs) for result in self.get_paged_data().results ] return self.get_paged_data().results @property def page_info(self): # FIXME Replace with a dataclass # return PageInfo(self.page_data.next_page_token) # FIXME, add more details like startCursor return { "hasNextPage": bool(self.get_paged_data().next_page_token), "endCursor": utils.get_encoded_cursor( self.get_paged_data().next_page_token ), } @property def total_count(self): """ Will be None for most of the connections override if needed/possible """ return None def get_paged_data(self): """ Cache to avoid multiple calls to the backend (_get_paged_result) return a PagedResult object """ if self._paged_data is None: # FIXME, make this call async (not for v1) self._paged_data = self._get_paged_result() return self._paged_data @abstractmethod def _get_paged_result(self): """ Override for desired behaviour return a PagedResult object """ # FIXME, make this call async (not for v1) return None def _get_edges(self): # FIXME, make cursor work per item # Cursor can't be None here return [{"cursor": "dummy", "node": node} for node in self.nodes] def _get_after_arg(self): """ Return the decoded next page token override to use a specific token """ return utils.get_decoded_cursor(self.kwargs.get("after")) def _get_first_arg(self): """ page_size is set to 50 by default """ return self.kwargs.get("first", self._page_size) diff --git a/swh/graphql/resolvers/base_node.py b/swh/graphql/resolvers/base_node.py index 43e8ed4..fe5cc40 100644 --- a/swh/graphql/resolvers/base_node.py +++ b/swh/graphql/resolvers/base_node.py @@ -1,46 +1,50 @@ -""" -""" - from abc import ABC, abstractmethod from collections import namedtuple class BaseNode(ABC): + """ + Base class for all the Node resolvers + """ def __init__(self, obj, info, node_data=None, **kwargs): self.obj = obj self.info = info self.kwargs = kwargs self._set_node(node_data) def _set_node(self, node_data): if node_data is None: node_data = self._get_node_data() self._node = self._get_node_from_data(node_data) def _get_node_from_data(self, node_data): + """ + Create an object from the dict + Override to support complex data structures + """ if type(node_data) is dict: return namedtuple("NodeObj", node_data.keys())(*node_data.values()) return node_data def __call__(self, *args, **kw): - """ - """ return self @abstractmethod def _get_node_data(self): """ Override for desired behaviour + This will be called only when + node_data is not available """ # FIXME, make this call async (not for v1) return None def __getattr__(self, name): """ Any property defined in the sub-class will get precedence over the _node attributes """ return getattr(self._node, name) def is_type_of(self): return self.__class__.__name__