diff --git a/swh/graphql/resolvers/base_connection.py b/swh/graphql/resolvers/base_connection.py index dc71e10..205a1e2 100644 --- a/swh/graphql/resolvers/base_connection.py +++ b/swh/graphql/resolvers/base_connection.py @@ -1,112 +1,112 @@ """ """ - 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): - _model_class = None + _model_class: Any = None def __init__(self, obj, info, **kwargs): self.obj = obj self.info = info self.kwargs = kwargs self._page_data = None self.pageInfo = self.page_info self.totalCount = self.total_count def __call__(self): return self @property def edges(self): return self._get_edges() @property def nodes(self): """ Override if needed return a list of objects If a model class is set, return a list of its intance else a list of nodes """ if self._model_class is not None: return [self._model_class(obj) for obj in self.page_data.results] return self.page_data.results @property def page_info(self): # FIXME Replace with a dataclass # return PageInfo(self.page_data.next_page_token) return { "hasNextPage": bool(self.page_data.next_page_token), "endCursor": utils.get_encoded_cursor(self.page_data.next_page_token), } @property def total_count(self): """ Will be None for most of the connections override if needed """ return None @property def page_data(self): """ Cache to avoid multiple calls to the backend """ if self._page_data is None: # FIXME, make this call async (not for v1) self._page_data = self._get_page_result() return self._page_data @abstractmethod def _get_page_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 return [{"cursor": "test", "node": each} for each 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): """ Override to set the default page size """ return self.kwargs.get("first", 50) diff --git a/swh/graphql/resolvers/base_node.py b/swh/graphql/resolvers/base_node.py index 408e2d3..d83780a 100644 --- a/swh/graphql/resolvers/base_node.py +++ b/swh/graphql/resolvers/base_node.py @@ -1,39 +1,40 @@ """ """ from abc import ABC, abstractmethod +from typing import Any class BaseNode(ABC): - _model_class = None + _model_class: Any = None def __init__(self, obj, info, **kwargs): self.obj = obj self.info = info self.kwargs = kwargs self._node = None def __call__(self): """ If a model class is set, return its instance, else node as it is """ if self._model_class is not None: return self._model_class(self.node) return self.node @property def node(self): # This is a small cache to avoid multiple # backend calls if self._node is None: self._node = self._get_node() return self._node @abstractmethod def _get_node(self): """ Override for desired behaviour """ # FIXME, make this call async (not for v1) return None