From 5a25ee85eb3c299b7625cee16af70319eea13a5e Mon Sep 17 00:00:00 2001 From: mwallschlaeger Date: Wed, 23 Aug 2023 17:05:50 +0200 Subject: [PATCH] issue#23_listing_exectionrequests_returns_error --- src/cmdprint.py | 41 ++++++++++++++++++++++++++++++++- src/datasets.py | 4 ++-- src/documents.py | 4 ++-- src/executionrequest.py | 17 +++++++------- src/geonodeobject.py | 50 +++++------------------------------------ src/people.py | 6 ++--- 6 files changed, 61 insertions(+), 61 deletions(-) diff --git a/src/cmdprint.py b/src/cmdprint.py index a6087ec..c669389 100644 --- a/src/cmdprint.py +++ b/src/cmdprint.py @@ -1,5 +1,8 @@ -from typing import List +from typing import List, Union, Dict from tabulate import tabulate +import json + +from src.geonodetypes import GeonodeCmdOutObjectKey def show_list(headers: List[str], values: List[List[str]], tablefmt="github"): @@ -11,3 +14,39 @@ def show_list(headers: List[str], values: List[List[str]], tablefmt="github"): tablefmt (str, optional): used tabulate table format, see https://pypi.org/project/tabulate/ """ print(tabulate(values, headers=headers, tablefmt=tablefmt)) + + +def __cmd_list_header__(cmdout_header: List[GeonodeCmdOutObjectKey]) -> List[str]: + """returns the default header to print list on cmd + + Returns: + List[str]: list of header elements as str + """ + return [str(cmdoutkey) for cmdoutkey in cmdout_header] + + +def print_list_on_cmd(obj: Dict, cmdout_header: List[GeonodeCmdOutObjectKey]): + """print a beautiful list on the cmdline + + Args: + obj (Dict): dict object to print on cmd line + """ + + def generate_line(i, obj: Dict, headers: List[GeonodeCmdOutObjectKey]) -> List: + return [cmdoutkey.get_key(obj[i]) for cmdoutkey in headers] + + values = [generate_line(i, obj, cmdout_header) for i in range(len(obj))] + show_list(headers=__cmd_list_header__(cmdout_header), values=values) + + +def print_json(json_str: Union[str, dict]): + """ + Print the given JSON string or dictionary with an indentation of 2 spaces. + + Args: + json_str (Union[str, dict]): The JSON string or dictionary to be printed. + + Returns: + None + """ + print(json.dumps(json_str, indent=2)) diff --git a/src/datasets.py b/src/datasets.py index 88959f2..5c58adf 100644 --- a/src/datasets.py +++ b/src/datasets.py @@ -4,7 +4,7 @@ from src.resources import GeonodeResourceHandler from src.geonodetypes import GeonodeHTTPFile -from src.cmdprint import show_list +from src.cmdprint import show_list, print_json from src.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey from src.executionrequest import GeonodeExecutionRequestHandler @@ -58,7 +58,7 @@ def cmd_upload( er = execution_request_handler.get(exec_id=str(r["execution_id"]), **kwargs) if kwargs["json"] is True: - self.print_json(er) + print_json(er) else: list_items = [ ["exec_id", str(er["exec_id"])], diff --git a/src/documents.py b/src/documents.py index a9ca138..7e3714b 100644 --- a/src/documents.py +++ b/src/documents.py @@ -4,7 +4,7 @@ from typing import List, Optional, Dict from src.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey -from src.cmdprint import show_list +from src.cmdprint import show_list, print_json from src.resources import GeonodeResourceHandler from src.geonodetypes import GeonodeHTTPFile @@ -50,7 +50,7 @@ def cmd_upload( ["download-url", r["href"]], ] if kwargs["json"]: - self.print_json(r) + print_json(r) else: show_list(values=list_items, headers=["key", "value"]) diff --git a/src/executionrequest.py b/src/executionrequest.py index 4cdba00..d9c41ab 100644 --- a/src/executionrequest.py +++ b/src/executionrequest.py @@ -1,8 +1,8 @@ -from src.geonodetypes import GeonodeCmdOutListKey -from src.geonodeobject import GeonodeObjectHandler -from src.rest import GeonodeRest +from typing import Dict, List -from typing import Dict +from src.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutObjectKey +from src.rest import GeonodeRest +from src.cmdprint import print_list_on_cmd, print_json class GeonodeExecutionRequestHandler(GeonodeRest): @@ -10,8 +10,7 @@ class GeonodeExecutionRequestHandler(GeonodeRest): JSON_OBJECT_NAME = "requests" SINGULAR_RESOURCE_NAME = "request" - # TODO - LIST_CMDOUT_HEADER = [ + LIST_CMDOUT_HEADER: List[GeonodeCmdOutObjectKey] = [ GeonodeCmdOutListKey(key="exec_id"), GeonodeCmdOutListKey(key="name"), GeonodeCmdOutListKey(key="status"), @@ -23,7 +22,7 @@ class GeonodeExecutionRequestHandler(GeonodeRest): def cmd_describe(self, exec_id: str, **kwargs): obj = self.get(exec_id=exec_id, **kwargs) - GeonodeObjectHandler.print_json(obj) + print_json(obj) def get(self, exec_id: str, **kwargs) -> Dict: """ @@ -44,9 +43,9 @@ def cmd_list(self, **kwargs): """show list of geonode obj on the cmdline""" obj = self.list(**kwargs) if kwargs["json"]: - GeonodeObjectHandler.print_json(obj) + print_json(obj) else: - GeonodeObjectHandler.print_list_on_cmd(obj) + print_list_on_cmd(obj, self.LIST_CMDOUT_HEADER) def list(self, **kwargs) -> Dict: """returns dict of datasets from geonode diff --git a/src/geonodeobject.py b/src/geonodeobject.py index eea0e70..e7e0179 100644 --- a/src/geonodeobject.py +++ b/src/geonodeobject.py @@ -1,9 +1,9 @@ -from typing import List, Dict, Union +from typing import List, Dict import json from src.geonodetypes import GeonodeCmdOutObjectKey, GeonodeCmdOutListKey from src.rest import GeonodeRest -from src.cmdprint import show_list +from src.cmdprint import print_list_on_cmd, print_json class GeonodeObjectHandler(GeonodeRest): @@ -22,9 +22,9 @@ def cmd_list(self, **kwargs): """show list of geonode obj on the cmdline""" obj = self.list(**kwargs) if kwargs["json"]: - self.print_json(obj) + print_json(obj) else: - self.__class__.print_list_on_cmd(obj) + print_list_on_cmd(obj, self.LIST_CMDOUT_HEADER) def list(self, **kwargs) -> Dict: """returns dict of datasets from geonode @@ -48,7 +48,7 @@ def delete(self, pk: int, **kwargs): def cmd_patch(self, pk: int, fields: str, **kwargs): obj = self.patch(pk, fields, **kwargs) - self.print_json(obj) + print_json(obj) def patch(self, pk: int, fields: str, **kwargs) -> Dict: """tries to generate object from incoming json string @@ -76,7 +76,7 @@ def patch(self, pk: int, fields: str, **kwargs) -> Dict: def cmd_describe(self, pk: int, **kwargs): obj = self.get(pk=pk, **kwargs) - self.print_json(obj) + print_json(obj) def get(self, pk: int, **kwargs) -> Dict: """get details for a given pk @@ -91,41 +91,3 @@ def get(self, pk: int, **kwargs) -> Dict: endpoint=f"{self.ENDPOINT_NAME}/{pk}?page_size={kwargs['page_size']}" ) return r[self.SINGULAR_RESOURCE_NAME] - - @classmethod - def cmd_list_header(self) -> List[str]: - """returns the default header to print list on cmd - - Returns: - List[str]: list of header elements as str - """ - return [str(cmdoutkey) for cmdoutkey in self.LIST_CMDOUT_HEADER] - - @classmethod - def print_list_on_cmd(cls, obj: Dict): - """print a beautiful list on the cmdline - - Args: - obj (Dict): dict object to print on cmd line - """ - - def generate_line(i, obj: Dict, headers: List[GeonodeCmdOutObjectKey]) -> List: - return [cmdoutkey.get_key(obj[i]) for cmdoutkey in headers] - - values = [ - generate_line(i, obj, cls.LIST_CMDOUT_HEADER) for i in range(len(obj)) - ] - show_list(headers=cls.cmd_list_header(), values=values) - - @classmethod - def print_json(cls, json_str: Union[str, dict]): - """ - Print the given JSON string or dictionary with an indentation of 2 spaces. - - Args: - json_str (Union[str, dict]): The JSON string or dictionary to be printed. - - Returns: - None - """ - print(json.dumps(json_str, indent=2)) diff --git a/src/people.py b/src/people.py index cb7c993..f657299 100644 --- a/src/people.py +++ b/src/people.py @@ -1,8 +1,8 @@ from typing import Dict from src.geonodeobject import GeonodeObjectHandler -from src.resources import GeonodeResourceHandler from src.geonodetypes import GeonodeCmdOutListKey +from src.cmdprint import print_list_on_cmd, print_json class GeonodePeopleHandler(GeonodeObjectHandler): @@ -41,9 +41,9 @@ def cmd_describe( ) # in this case print as list of ressources if user_resources is True: - GeonodeResourceHandler.print_list_on_cmd(obj) + print_list_on_cmd(obj, self.LIST_CMDOUT_HEADER) else: - self.print_json(obj) + print_json(obj) def get( self, pk: int, user_resources: bool = False, user_groups: bool = False, **kwargs