Merge pull request #86 from GeoNodeUserGroup-DE/issue_#78_Feature_sup…

…port_/api/v2/resources/pk/linked_resources

Issue #78 feature support /api/v2/resources/pk/linked resources

.github/workflows/python-mypy.yaml

@@ -30,4 +30,4 @@ jobs:
           python -m pip install --upgrade pip mypy==1.10.0
           pip install -e .
       - name: Test with mypy
-        run: "mypy -p src.geonoderest --check-untyped-defs"
+        run: "mypy -p src.geonoderest --check-untyped-defs --disable-error-code=import-untyped"

pyproject.toml

@@ -26,3 +26,10 @@ repository = "https://github.com/GeoNodeUserGroup-DE/geonodectl/"
 [project.scripts]
 geonodectl = "geonoderest.geonodectl:geonodectl"
 
+[tool.pytest.ini_options]
+minversion = "6.0"
+addopts = "-ra -q"
+testpaths = [
+    "src/geonoderest/tests/",
+
+]

src/geonoderest/cmdprint.py

@@ -51,6 +51,9 @@ def print_json(json_str: Union[str, dict]):
     Returns:
         None
     """
+    if json_str is None:
+        logging.warning("return from geonode api was broken, not output ...")
+        return None
     print(json.dumps(json_str, indent=2))
 
 

src/geonoderest/datasets.py

@@ -1,12 +1,13 @@
 import os
 from pathlib import Path
 from typing import List, Dict
+import logging
 
-from .resources import GeonodeResourceHandler
-from .geonodetypes import GeonodeHTTPFile
-from .cmdprint import show_list, print_json
-from .geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
-from .executionrequest import GeonodeExecutionRequestHandler
+from geonoderest.resources import GeonodeResourceHandler
+from geonoderest.geonodetypes import GeonodeHTTPFile
+from geonoderest.cmdprint import show_list, print_json
+from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
+from geonoderest.executionrequest import GeonodeExecutionRequestHandler
 
 
 class GeonodeDatasetsHandler(GeonodeResourceHandler):
@@ -56,7 +57,9 @@ def cmd_upload(
             env=self.gn_credentials
         )
         er = execution_request_handler.get(exec_id=str(r["execution_id"]), **kwargs)
-
+        if er is None:
+            logging.warning("upload failed ... ")
+            return
         if kwargs["json"] is True:
             print_json(er)
         else:
@@ -149,7 +152,7 @@ def upload(
                     ("zip_file", (dataset_path.name, open(dataset_path, "rb")))
                 )
 
-        params = {
+        json = {
             # layer permissions
             "permissions": '{ "users": {"AnonymousUser": ["view_resourcebase"]} , "groups":{}}',
             "mosaic": mosaic,
@@ -161,6 +164,6 @@ def upload(
         return self.http_post(
             endpoint="uploads/upload",
             files=files,
-            params=params,
+            json=json,
             content_length=content_length,
         )

src/geonoderest/documents.py

@@ -2,11 +2,12 @@
 import mimetypes
 from pathlib import Path
 from typing import List, Optional, Dict
+import logging
 
-from .geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
-from .cmdprint import show_list, print_json
-from .resources import GeonodeResourceHandler
-from .geonodetypes import GeonodeHTTPFile
+from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
+from geonoderest.cmdprint import show_list, print_json
+from geonoderest.resources import GeonodeResourceHandler
+from geonoderest.geonodetypes import GeonodeHTTPFile
 
 
 class GeonodeDocumentsHandler(GeonodeResourceHandler):
@@ -41,6 +42,10 @@ def cmd_upload(
         r = self.upload(
             file_path=file_path, metadata_only=metadata_only, charset=charset, **kwargs
         )
+        if r is None:
+            logging.warning("upload failed ... ")
+            return
+
         list_items = [
             ["name", r["title"]],
             ["state", r["state"]],
@@ -61,7 +66,7 @@ def upload(
         charset: str = "UTF-8",
         metadata_only: bool = False,
         **kwargs,
-    ) -> Dict:
+    ) -> Optional[Dict]:
         """upload a document to geonode
 
         Args:
@@ -73,14 +78,14 @@ def upload(
             FileNotFoundError: raises file not found if the given filepath is not accessable
 
         Returns:
-            Dict: returns json response from geonode as dict
+            Optional[Dict]: returns json response from geonode as dict
         """
 
         document_path: Path = file_path
         if not document_path.exists():
             raise FileNotFoundError
 
-        params = {
+        json = {
             # layer permissions
             "permissions": '{ "users": {"AnonymousUser": ["view_resourcebase"]} , "groups":{}}',
             "charset": charset,
@@ -102,7 +107,9 @@ def upload(
         r = self.http_post(
             endpoint="documents",
             files=files,
-            params=params,
+            json=json,
             content_length=content_length,
-        )["document"]
-        return r
+        )
+        if r is None:
+            return None
+        return r["document"]

src/geonoderest/exceptions.py

@@ -0,0 +1,7 @@
+class GeoNodeRestException(Exception):
+    """
+    GeoNodeRestException
+
+    """
+
+    pass

src/geonoderest/executionrequest.py

@@ -1,9 +1,10 @@
-from typing import Dict, List
+from typing import Dict, List, Optional
+import logging
 
-from .geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutObjectKey
-from .rest import GeonodeRest
+from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutObjectKey
+from geonoderest.rest import GeonodeRest
 
-from .cmdprint import print_list_on_cmd, print_json
+from geonoderest.cmdprint import print_list_on_cmd, print_json
 
 
 class GeonodeExecutionRequestHandler(GeonodeRest):
@@ -41,12 +42,15 @@ def get(self, exec_id: str, **kwargs) -> Dict:
     def cmd_list(self, **kwargs):
         """show list of geonode obj on the cmdline"""
         obj = self.list(**kwargs)
+        if obj is None:
+            logging.warning("getting list failed ...")
+            return None
         if kwargs["json"]:
             print_json(obj)
         else:
             print_list_on_cmd(obj, self.LIST_CMDOUT_HEADER)
 
-    def list(self, **kwargs) -> Dict:
+    def list(self, **kwargs) -> Optional[Dict]:
         """returns dict of execution requests from geonode
 
         Returns:
@@ -56,4 +60,6 @@ def list(self, **kwargs) -> Dict:
 
         params = self.__handle_http_params__({}, kwargs)
         r = self.http_get(endpoint=endpoint, params=params)
+        if r is None:
+            return None
         return r[self.JSON_OBJECT_NAME]

src/geonoderest/geoapps.py

@@ -1,5 +1,5 @@
-from .resources import GeonodeResourceHandler
-from .geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
+from geonoderest.resources import GeonodeResourceHandler
+from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
 
 
 class GeonodeGeoappsHandler(GeonodeResourceHandler):