Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Fixes #23] Listing exectionrequests returns error #24

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 40 additions & 1 deletion src/cmdprint.py
Original file line number Diff line number Diff line change
@@ -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"):
Expand All @@ -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))
4 changes: 2 additions & 2 deletions src/datasets.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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"])],
Expand Down
4 changes: 2 additions & 2 deletions src/documents.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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"])
Expand Down
17 changes: 8 additions & 9 deletions src/executionrequest.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,16 @@
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):
ENDPOINT_NAME = "executionrequest"
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"),
Expand All @@ -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:
"""
Expand All @@ -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
Expand Down
50 changes: 6 additions & 44 deletions src/geonodeobject.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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))
6 changes: 3 additions & 3 deletions src/people.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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
Expand Down