Constants and Paths

lib/StaticNarrative/__init__.py

@@ -1,6 +1,6 @@
-import os
+"""StaticNarrative directory."""
+
+from pathlib import Path
 
 # Set a variable to refer to the base directory of this repo
-STATIC_NARRATIVE_BASE_DIR = os.path.dirname(
-    os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-)
+STATIC_NARRATIVE_BASE_DIR = Path(__file__).resolve().parents[2]

lib/StaticNarrative/config.py

@@ -1,6 +1,7 @@
 """Config for the StaticNarrative app."""
 
 import os
+from pathlib import Path
 from typing import Any
 from urllib.parse import urlparse
 
@@ -38,18 +39,21 @@ def generate_config(config: dict[str, Any] | None) -> dict[str, Any]:
 
     # ensure these paths are absolute, not relative
     for path in ["static_file_root", "scratch"]:
-        assigned_path = config.get(path)
+        # we have already checked that the path is not None
+        assigned_path = Path(config.get(path))
 
-        if not os.path.isabs(assigned_path):
-            config[path] = os.path.abspath(assigned_path)
+        if not assigned_path.is_absolute():
+            assigned_path = assigned_path.resolve()
 
         # check that the directory exists and is writable
-        if not os.path.isdir(config[path]):
+        if not assigned_path.is_dir():
             msg = f"{path}: {config[path]} is not a directory"
             raise RuntimeError(msg)
 
-        if path == "scratch" and not os.access(config[path], os.W_OK):
-            msg = f"Cannot write to directory {config[path]}"
+        if path == "scratch" and not os.access(assigned_path, os.W_OK):
+            msg = f"Cannot write to directory {assigned_path}"
             raise RuntimeError(msg)
 
+        config[path] = str(assigned_path)
+
     return config

lib/StaticNarrative/constants.py

@@ -0,0 +1,14 @@
+"""Constants used throughout the codebase."""
+
+DATA = "data"
+ID = "by_id"
+INFO = "info"
+INFOSTR = "infostruct"
+NARRATIVE_TEMPLATE_FILE = "narrative.tpl"
+NARRATIVE_TYPE = "KBaseNarrative.Narrative"
+OBJ_INFO = "object_info"
+OUTPUT_DATA_FILE = "data.json"
+OUTPUT_HTML_FILE = "narrative.html"
+REPORT_TYPE = "KBaseReport."
+SAVED_HTML_FILE = "index.html"
+TYPE = "by_type"

lib/StaticNarrative/creator.py

@@ -1,11 +1,12 @@
 """Class for creating static narratives."""
 
 import logging
-import os
+from pathlib import Path
 from typing import Any
 
 from installed_clients.WorkspaceClient import Workspace
 
+from StaticNarrative.constants import NARRATIVE_TYPE
 from StaticNarrative.exporter.exporter import NarrativeExporter
 from StaticNarrative.narrative.narrative_util import (
     save_narrative_url,
@@ -31,8 +32,8 @@
         :return: nothing
         :rtype: None
         """
-        if not token or not config["workspace_url"]:
-            msg = "workspace URL and a token required to initialise the StaticNarrativeCreator."
+        if not token or not config or "workspace_url" not in config:
+            msg = "Workspace URL and a token required to initialise the StaticNarrativeCreator."
             raise RuntimeError(msg)
 
         self.config = config
@@ -48,62 +49,6 @@
         ch.setFormatter(formatter)
         self.logger.addHandler(ch)
 
-    def get_narrative_id_from_workspace(self: "StaticNarrativeCreator", ws_ref: str | int) -> str:
-        """Retrieve the Narrative object from a workspace (if one exists).
-
-        :param self: this class
-        :type self: StaticNarrativeCmdLine
-        :param ws_ref: workspace reference (ID or name)
-        :type ws_ref: str | int
-        :raises ValueError: if no Narrative object is found
-        :return: KBaseNarrative.Narrative object UPA
-        :rtype: str
-        """
-        ws_args = "workspaces"
-        if str(ws_ref).isdigit():
-            ws_args = "ids"
-
-        results = self.ws_client.list_objects(
-            {ws_args: [ws_ref], "type": "KBaseNarrative.Narrative"}
-        )
-        if results and results[0] and results[0][0]:
-            return f"{results[0][6]}/{results[0][0]}/{results[0][4]}"
-
-        # no narrative object
-        msg = f"Workspace {ws_ref} did not contain a KBaseNarrative.Narrative object."
-        raise ValueError(msg)
-
-    def create_local_static_narrative(
-        self: "StaticNarrativeCreator",
-        ws_ref: str | int,
-        user_id: str,
-        skip_permissions_checks: str,
-    ) -> None:
-        """Create a static narrative from a Workspace reference.
-
-        :param ws_ref: workspace reference; may be the narrative UPA or a workspace ID
-        :type ws_ref: str
-        :param user_id: valid KBase user ID
-        :type user_id: str
-        :param skip_permissions_checks: whether the permission checks should be run
-        :type skip_permissions_checks: str
-        """
-        # this is a workspace reference
-        ws_ref = str(ws_ref)
-        narrative_ref = ws_ref
-        if ws_ref.count("/") == 0:
-            narrative_ref = self.get_narrative_id_from_workspace(ws_ref)
-
-        ref = NarrativeRef.parse(narrative_ref)
-
-        self.logger.info("Creating Static Narrative %s", ref)
-
-        if not skip_permissions_checks:
-            self.check_permissions(ref, user_id=user_id)
-
-        output_path = self.export_narrative(ref, user_id)
-        self.logger.info("Static Narrative for %s created at %s", ref, output_path)
-
     def create_static_narrative(
         self: "StaticNarrativeCreator", params: dict[str, str]
     ) -> dict[str, str]:
@@ -121,7 +66,7 @@
         self.check_permissions(ref, user_id=user_id)
         output_path = self.export_narrative(ref, user_id=user_id)
         # get the output directory for the upload_and_save command
-        output_dir = os.path.dirname(output_path)
+        output_dir = output_path.parent
         static_url = self.upload_and_save(ref, output_dir)
 
         return {"static_narrative_url": static_url}
@@ -137,44 +82,47 @@
         verify_admin_privilege(self.ws_client, user_id, ref.wsid)
         verify_public_narrative(self.ws_client, ref.wsid)
 
-    def export_narrative(self: "StaticNarrativeCreator", ref: NarrativeRef, user_id: str) -> str:
+    def export_narrative(self: "StaticNarrativeCreator", ref: NarrativeRef, user_id: str) -> Path:
         """Create an output directory and export the SN to a file.
 
         :param ref: reference for the narrative
         :type ref: NarrativeRef
         :param user_id: user ID
         :type user_id: str
         :return: path to the SN created by the exporter
-        :rtype: str
+        :rtype: Path
         """
         exporter = NarrativeExporter(self.config, user_id, self.token)
 
         # set up output directories
         try:
-            output_dir = os.path.join(
-                self.config["scratch"], str(ref.wsid), str(ref.objid), str(ref.ver)
+            output_dir = (
+                Path(self.config["scratch"]) / str(ref.wsid) / str(ref.objid) / str(ref.ver)
             )
-            os.makedirs(output_dir, exist_ok=True)
+            output_dir.mkdir(exist_ok=True, parents=True)
         except OSError as e:
-            self.logger.exception("Error while creating Static Narrative directory: ", e)
-            raise
+            self.logger.exception("Error while creating Static Narrative directory")
+            msg = "Could not create static narrative directory"
+            raise RuntimeError(msg) from e
 
         # export the narrative to a file
         try:
             output_path = exporter.export_narrative(ref, output_dir)
-        except Exception as e:
-            self.logger.exception("Error while exporting Narrative", e)
+        except Exception:
+            self.logger.exception("Error while exporting Narrative")
             raise
 
         return output_path
 
-    def upload_and_save(self: "StaticNarrativeCreator", ref: NarrativeRef, output_path: str) -> str:
+    def upload_and_save(
+        self: "StaticNarrativeCreator", ref: NarrativeRef, output_path: Path
+    ) -> str:
         """Upload the static narrative and save the URL to the ws metadata.
 
         :param ref: reference for the narrative
         :type ref: NarrativeRef
         :param output_path: path to the saved output
-        :type output_path: str
+        :type output_path: Path
         :return: URL for the static narrative
         :rtype: str
         """
@@ -183,3 +131,61 @@
         save_narrative_url(self.ws_client, ref, static_url)
         self.logger.info("Finished creating Static Narrative %s", ref)
         return static_url
+
+    def get_narrative_id_from_workspace(self: "StaticNarrativeCreator", ws_ref: str | int) -> str:
+        """Retrieve the Narrative object from a workspace (if one exists).
+
+        This function is used when creating a Static Narrative locally using a workspace ID as input.
+
+        :param self: this class
+        :type self: StaticNarrativeCmdLine
+        :param ws_ref: workspace reference (ID or name)
+        :type ws_ref: str | int
+        :raises ValueError: if no Narrative object is found
+        :return: KBaseNarrative.Narrative object UPA
+        :rtype: str
+        """
+        ws_args = "workspaces"
+        if str(ws_ref).isdigit():
+            ws_args = "ids"
+
+        results = self.ws_client.list_objects({ws_args: [ws_ref], "type": NARRATIVE_TYPE})
+        if results and results[0] and results[0][0]:
+            return f"{results[0][6]}/{results[0][0]}/{results[0][4]}"
+
+        # no narrative object
+        msg = f"Workspace {ws_ref} did not contain a {NARRATIVE_TYPE} object."
+        raise ValueError(msg)
+
+    def create_local_static_narrative(
+        self: "StaticNarrativeCreator",
+        ws_ref: str | int,
+        user_id: str,
+        skip_permissions_checks: str,
+    ) -> None:
+        """Create a static narrative from a Workspace reference.
+
+        This function is used when creating a Static Narrative locally using a workspace ID as input.
+
+        :param ws_ref: workspace reference; may be the narrative UPA or a workspace ID
+        :type ws_ref: str
+        :param user_id: valid KBase user ID
+        :type user_id: str
+        :param skip_permissions_checks: whether the permission checks should be run
+        :type skip_permissions_checks: str
+        """
+        # this is a workspace reference
+        ws_ref = str(ws_ref)
+        narrative_ref = ws_ref
+        if ws_ref.count("/") == 0:
+            narrative_ref = self.get_narrative_id_from_workspace(ws_ref)
+
+        ref = NarrativeRef.parse(narrative_ref)
+
+        self.logger.info("Creating Static Narrative %s", ref)
+
+        if not skip_permissions_checks:
+            self.check_permissions(ref, user_id=user_id)
+
+        output_path = self.export_narrative(ref, user_id)
+        self.logger.info("Static Narrative for %s created at %s", ref, str(output_path))

lib/StaticNarrative/exceptions.py

@@ -1,3 +1,5 @@
+"""Custom StaticNarrative Errors."""
+
 import re
 from typing import Any
 
@@ -9,7 +11,9 @@ class PermissionsError(ServerError):
 
     @staticmethod
     def is_permissions_error(err: str) -> bool:
-        """Try to guess if the error string is a permission-denied error
+        """Parse the error string for permissions-related text.
+
+        Try to guess if the error string is a permission-denied error
         for the narrative (i.e. the workspace the narrative is in).
         """
         pat = re.compile(
@@ -25,18 +29,23 @@ def __init__(
         message: str | None = None,
         **kw: None | dict[str, Any],
     ) -> None:
+        """Create a PermissionsError instance."""
         ServerError.__init__(self, name, code, message, **kw)
 
 
 class WorkspaceError(Exception):
+    """Raised if an error occurs during a workspace call."""
+
     def __init__(
         self: "WorkspaceError",
         ws_server_err: ServerError,
         ws_id: str | int,
         message: str | None = None,
         http_code: int | None = 500,
     ) -> None:
-        """This wraps Workspace calls regarding Narratives into exceptions that are
+        """An error generated by a workspace call.
+
+        This wraps Workspace calls regarding Narratives into exceptions that are
         easier to parse for logging, user communication, etc.
 
         ws_server_err should be the ServerError that comes back from a workspace
@@ -67,4 +76,5 @@ def __init__(
             self.message = ws_server_err.message
 
     def __str__(self: "WorkspaceError") -> str:
+        """Return the error as a string."""
         return f"WorkspaceError: {self.ws_id}: {self.http_code}: {self.message}"

lib/StaticNarrative/exporter/data_exporter.py

@@ -6,13 +6,15 @@
 
 from installed_clients.WorkspaceClient import Workspace
 
+from StaticNarrative.constants import NARRATIVE_TYPE, OBJ_INFO, OUTPUT_DATA_FILE
 from StaticNarrative.exporter.dynamic_service_client import DynamicServiceClient
+from StaticNarrative.exporter.icon_util import get_data_icon
 from StaticNarrative.exporter.objects_with_sets import ObjectsWithSets
-from StaticNarrative.exporter.processor_util import get_data_icon
 from StaticNarrative.upa import generate_upa
 
-IGNORED_TYPES = ["KBaseNarrative.Narrative"]
-OUTPUT_DATA_FILE = "data.json"
+IGNORED_TYPES = [NARRATIVE_TYPE]
+SET_ITEMS = "set_items"
+SET_ITEMS_INFO = "set_items_info"
 
 
 def export_narrative_data(
@@ -60,24 +62,24 @@ def export_narrative_data(
     indexed_data = {}
     type_info = {}
     for item in ws_data:
-        obj = item["object_info"]
+        obj = item[OBJ_INFO]
         obj_type = obj[2].split("-")[0]
         if obj_type in IGNORED_TYPES:
             continue
-        obj_upa = generate_upa(item["object_info"])
+        obj_upa = generate_upa(item[OBJ_INFO])
         type_name = obj_type.split(".")[-1]
         if type_name not in type_info:
             type_info[type_name] = {"count": 0, "icon": get_data_icon(type_name)}
         filtered_data.append(_reshape_obj(obj, obj_upa))
         type_info[type_name]["count"] += 1
 
         # if this is a set, go through each item in the set and add it to indexed_data
-        if "set_items" in item and "set_items_info" in item["set_items"]:
-            item["set_items"]["upas"] = []
-            for set_item in item["set_items"]["set_items_info"]:
+        if SET_ITEMS in item and SET_ITEMS_INFO in item[SET_ITEMS]:
+            item[SET_ITEMS]["upas"] = []
+            for set_item in item[SET_ITEMS][SET_ITEMS_INFO]:
                 set_item_upa = generate_upa(set_item)
-                indexed_data[set_item_upa] = {"object_info": set_item}
-                item["set_items"]["upas"].append(set_item_upa)
+                indexed_data[set_item_upa] = {OBJ_INFO: set_item}
+                item[SET_ITEMS]["upas"].append(set_item_upa)
 
         # add the item to the index of ws objects
         indexed_data[obj_upa] = item
@@ -89,7 +91,7 @@ def export_narrative_data(
     }
 
     output_path = Path(output_dir) / OUTPUT_DATA_FILE
-    with open(output_path, "w") as outfile:
+    with output_path.open("w") as outfile:
         json.dump(output_data, outfile)
     output_data["path"] = str(output_path)
     output_data["indexed_data"] = indexed_data