Merge pull request #74 from dwolfson/v5.2

Addresses Issue: Separate path from file name in glossary/import and export #71

Author: dwolfson

pyegeria/commands/cat/glossary_actions.py

@@ -48,6 +48,7 @@
 EGERIA_WIDTH = os.environ.get("EGERIA_WIDTH", 200)
 EGERIA_JUPYTER = os.environ.get("EGERIA_JUPYTER", False)
 EGERIA_HOME_GLOSSARY_GUID = os.environ.get("EGERIA_HOME_GLOSSARY_GUID", None)
+EGERIA_GLOSSARY_PATH = os.environ.get("EGERIA_GLOSSARY_PATH", None)
 
 
 @click.command("create-glossary")
@@ -265,8 +266,11 @@ def delete_term(server, url, userid, password, timeout, term_guid):
 
 
 @click.command("import-terms")
-@click.option("--glossary-name", help="Name of Glossary", required=True)
-@click.option("--file-name", help="Path of CSV file", required=True)
+@click.option("--glossary_name", help="Name of Glossary", required=True)
+@click.option("--file_name", help="Path of CSV file", required=True)
+@click.option(
+    "--file_path", help="Path of CSV file", default=EGERIA_GLOSSARY_PATH, required=False
+)
 @click.option(
     "--verbose",
     is_flag=True,
@@ -288,6 +292,7 @@ def delete_term(server, url, userid, password, timeout, term_guid):
 @click.option("--timeout", default=60, help="Number of seconds to wait")
 def import_terms(
     glossary_name: str,
+    file_path: str,
     file_name: str,
     verbose: bool,
     upsert: bool,
@@ -302,7 +307,11 @@ def import_terms(
     token = m_client.create_egeria_bearer_token()
     try:
         result = m_client.load_terms_from_file(
-            glossary_name, file_name, upsert=upsert, verbose=verbose
+            glossary_name,
+            file_name,
+            file_path=file_path,
+            upsert=upsert,
+            verbose=verbose,
         )
 
         click.echo(
@@ -319,25 +328,30 @@ def import_terms(
 
 @click.command("export-terms")
 @click.option(
-    "--glossary-guid",
+    "--glossary_guid",
     default=EGERIA_HOME_GLOSSARY_GUID,
     help="GUID of Glossary to export",
     required=True,
 )
-@click.option("--file-name", help="Path of CSV file", required=True)
+@click.option("--file_name", help="Path of CSV file", required=True)
+@click.option(
+    "--file_path", help="Path of CSV file", default=EGERIA_GLOSSARY_PATH, required=False
+)
 @click.option("--server", default=EGERIA_VIEW_SERVER, help="Egeria view server to use")
 @click.option(
     "--url", default=EGERIA_VIEW_SERVER_URL, help="URL of Egeria platform to connect to"
 )
 @click.option("--userid", default=EGERIA_USER, help="Egeria user")
 @click.option("--password", default=EGERIA_USER_PASSWORD, help="Egeria user password")
 @click.option("--timeout", default=60, help="Number of seconds to wait")
-def export_terms(glossary_guid: str, file_name, server, url, userid, password, timeout):
+def export_terms(
+    glossary_guid: str, file_name, file_path, server, url, userid, password, timeout
+):
     """Export the glossary specified"""
     m_client = EgeriaTech(server, url, user_id=userid, user_pwd=password)
     token = m_client.create_egeria_bearer_token()
     try:
-        result = m_client.export_glossary_to_csv(glossary_guid, file_name)
+        result = m_client.export_glossary_to_csv(glossary_guid, file_name, file_path)
 
         click.echo(
             f"Exported {result} terms  from glossary: {glossary_guid} into {file_name}"

pyegeria/commands/cli/egeria.py

@@ -879,7 +879,7 @@ def glossary_group(ctx):
 )
 @click.option(
     "--glossary-guid",
-    default=None,
+    default=os.environ.get("EGERIA_HOME_GLOSSARY_GUID", None),
     help="Optionally restrict search to glossary with the specified guid",
 )
 @click.option(

pyegeria/commands/cli/egeria_cat.py

@@ -129,7 +129,8 @@
 )
 @click.option(
     "--width",
-    default=os.environ.get("EGERIA_WIDTH", "200"),
+    default=os.environ.get("EGERIA_WIDTH", 200),
+    type=int,
     help="Screen width, in characters, to use",
 )
 @click.option(
@@ -311,7 +312,7 @@ def glossary_group(ctx):
 )
 @click.option(
     "--glossary-guid",
-    default=os.environ.get("EGERIA_HOME_GLOSSARY_GUID"),
+    default=os.environ.get("EGERIA_HOME_GLOSSARY_GUID", None),
     help="Optionally restrict search to glossary with the specified guid",
 )
 @click.option(

pyegeria/glossary_manager_omvs.py

@@ -1535,6 +1535,7 @@ def load_terms_from_file(
         self,
         glossary_name: str,
         filename: str,
+        file_path: str = os.environ.get("EGERIA_GLOSSARY_PATH", None),
         upsert: bool = True,
         verbose: bool = True,
     ) -> List[dict] | None:
@@ -1544,6 +1545,9 @@ def load_terms_from_file(
         ----------
             glossary_name : str
                 Name of the glossary to import terms into.
+            file_path: str, default is EGERIA_GLOSSARY_PATH if specified or None
+                If EGERIA_GLOSSARY_PATH environment variable is set, then it will be used in forming the
+                prepended to the filename parameter to form the full path to the file.
             filename: str
                 Path to the file to import terms from. File is assumed to be in CSV format. The path
                 is relative to where the python method is being called from.
@@ -1609,16 +1613,25 @@ def load_terms_from_file(
             "Version Identifier",
             "Status",
         }
+
+        if file_path:
+            full_file_path = os.path.join(file_path, filename)
+        else:
+            full_file_path = filename
+
+        if not os.path.isfile(full_file_path):
+            raise FileNotFoundError(
+                f"Did not find file with path {file_path} and name {filename}"
+            )
         # process file
-        with open(filename, mode="r") as file:
+        with open(full_file_path, mode="r") as file:
             # Create a CSV reader object
             csv_reader = csv.DictReader(file)
             headers = csv_reader.fieldnames
             term_info = []
             # check that the column headers are known
             if all(header in term_properties for header in headers) is False:
                 raise InvalidParameterException("Invalid headers in CSV File")
-                sys.exit(1)
 
             # process each row and validate values
             for row in csv_reader:
@@ -1759,7 +1772,10 @@ def load_terms_from_file(
             return
 
     async def _async_export_glossary_to_csv(
-        self, glossary_guid: str, target_file: str
+        self,
+        glossary_guid: str,
+        target_file: str,
+        file_path: str = os.environ.get("EGERIA_GLOSSARY_PATH", None),
     ) -> int:
         """Export all the terms in a glossary to a CSV file. Async version
 
@@ -1769,6 +1785,9 @@ async def _async_export_glossary_to_csv(
             Identity of the glossary to export.
         target_file: str
             Complete file name with path and extension to export to.
+        file_path: str, default is EGERIA_GLOSSARY_PATH if specified or None
+                If EGERIA_GLOSSARY_PATH environment variable is set, then it will be used in forming the
+                prepended to the filename parameter to form the full path to the file.
 
         Returns:
             int: Number of rows exported.
@@ -1787,8 +1806,12 @@ async def _async_export_glossary_to_csv(
             "Version Identifier",
             "Status",
         ]
+        if file_path:
+            full_file_path = os.path.join(file_path, target_file)
+        else:
+            full_file_path = target_file
 
-        with open(target_file, mode="w") as file:
+        with open(full_file_path, mode="w") as file:
             csv_writer = csv.DictWriter(file, fieldnames=header)
             csv_writer.writeheader()
             count = 0
@@ -1822,7 +1845,12 @@ async def _async_export_glossary_to_csv(
                 count += 1
         return count
 
-    def export_glossary_to_csv(self, glossary_guid: str, target_file: str) -> int:
+    def export_glossary_to_csv(
+        self,
+        glossary_guid: str,
+        target_file: str,
+        file_path: str = os.environ.get("EGERIA_GLOSSARY_PATH", None),
+    ) -> int:
         """Export all the terms in a glossary to a CSV file.
 
         Parameters:
@@ -1831,14 +1859,17 @@ def export_glossary_to_csv(self, glossary_guid: str, target_file: str) -> int:
             Identity of the glossary to export.
         target_file: str
             Complete file name with path and extension to export to.
+        file_path: str, default is EGERIA_GLOSSARY_PATH if specified or None
+                If EGERIA_GLOSSARY_PATH environment variable is set, then it will be used in forming the
+                prepended to the filename parameter to form the full path to the file.
 
         Returns:
             int: Number of rows exported.
         """
 
         loop = asyncio.get_event_loop()
         response = loop.run_until_complete(
-            self._async_export_glossary_to_csv(glossary_guid, target_file)
+            self._async_export_glossary_to_csv(glossary_guid, target_file, file_path)
         )
 
         return response

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyegeria"
-version = "5.2.0.42.4"
+version = "5.2.0.42.8"
 license = 'Apache 2.0'
 authors = ["Dan Wolfson <[email protected]>"]
 readme = "README.md"

tests/test_glossary_manager_omvs.py

@@ -585,8 +585,9 @@ def test_load_terms_from_csv(self):
             )
             token = g_client.create_egeria_bearer_token(self.good_user_3, "secret")
             glossary = "example"
-            file_name = "/Users/dwolfson/localGit/egeria-v5-1/egeria-workspaces/exchange/loading-bay/glossary/upsert-example.om-terms"
-            response = g_client.load_terms_from_file(glossary, file_name, True)
+            file_path = "/Users/dwolfson/localGit/egeria-v5-1/egeria-workspaces/exchange/loading-bay/glossary"
+            file_name = "pets.om-terms"
+            response = g_client.load_terms_from_file(glossary, file_name, file_path)
             print(f"type is {type(response)}")
             if type(response) is list:
                 print("\n\n" + json.dumps(response, indent=4))
@@ -598,6 +599,7 @@ def test_load_terms_from_csv(self):
             InvalidParameterException,
             PropertyServerException,
             UserNotAuthorizedException,
+            FileNotFoundError,
         ) as e:
             print_exception_response(e)
             assert False, "Invalid request"