Creating cmf repo push and pull commands.

cmflib/cli/parser.py

@@ -17,11 +17,11 @@
 """Main parser for the cmf cli"""
 import argparse
 
-from cmflib.commands import artifact, metadata, init, execution, pipeline
+from cmflib.commands import artifact, metadata, init, execution, pipeline, repo
 
 from cmflib.cli import CmfParserError
 
-COMMANDS = [artifact, metadata, init, execution, pipeline]
+COMMANDS = [artifact, metadata, init, execution, pipeline, repo]
 
 
 def _find_parser(parser, cmd_cls):

cmflib/cmf.py

@@ -71,6 +71,8 @@
     _artifact_list,
     _pipeline_list,
     _execution_list,
+    _repo_push,
+    _repo_pull,
 )
 
 class Cmf:
@@ -1996,42 +1998,42 @@ def commit_existing(self, uri: str, props: t.Optional[t.Dict] = None, custom_pro
 #                print(last)
 #                os.symlink(str(index), slicedir + "/ " + last)
 
-def metadata_push(pipeline_name: str, filepath = "./mlmd", tensorboard_path: str = "", execution_id: str = ""):
+def metadata_push(pipeline_name: str, filepath = "./mlmd", tensorboard_path: str = "", execution_uuid: str = ""):
     """ Pushes MLMD file to CMF-server.
     Example:
     ```python
-         result = metadata_push("example_pipeline", "mlmd_file", "3")
+         result = metadata_push("example_pipeline", "mlmd_file", "eg_execution_uuid")
     ```
     Args:
         pipeline_name: Name of the pipeline.
         filepath: Path to the MLMD file.
-        execution_id: Optional execution ID.
+        execution_uuid: Optional execution UUID.
         tensorboard_path: Path to tensorboard logs.
 
     Returns:
         Response output from the _metadata_push function.
     """
     # Required arguments:  pipeline_name
-    # Optional arguments: Execution_ID, filepath (mlmd file path, tensorboard_path
-    output = _metadata_push(pipeline_name, filepath, execution_id, tensorboard_path)
+    # Optional arguments: Execution_UUID, filepath (mlmd file path), tensorboard_path
+    output = _metadata_push(pipeline_name, filepath, execution_uuid, tensorboard_path)
     return output
 
-def metadata_pull(pipeline_name: str, filepath = "./mlmd", execution_id: str = ""):
+def metadata_pull(pipeline_name: str, filepath = "./mlmd", execution_uuid: str = ""):
     """ Pulls MLMD file from CMF-server. 
      Example: 
      ```python 
-          result = metadata_pull("example_pipeline", "./mlmd_directory", "execution_123") 
+          result = metadata_pull("example_pipeline", "./mlmd_directory", "eg_execution_uuid") 
      ``` 
      Args: 
         pipeline_name: Name of the pipeline. 
         filepath: File path to store the MLMD file. 
-        execution_id: Optional execution ID. 
+        execution_uuid: Optional execution UUID. 
      Returns: 
         Message from the _metadata_pull function. 
      """
     # Required arguments:  pipeline_name 
-    #Optional arguments: Execution_ID, filepath(file path to store mlmd file) 
-    output = _metadata_pull(pipeline_name, filepath, execution_id)
+    #Optional arguments: Execution_UUID, filepath(file path to store mlmd file) 
+    output = _metadata_pull(pipeline_name, filepath, execution_uuid)
     return output
 
 def metadata_export(pipeline_name: str, jsonfilepath: str = "", filepath = "./mlmd"):
@@ -2331,23 +2333,23 @@ def pipeline_list(filepath = "./mlmd"):
     return output
 
 
-def execution_list(pipeline_name: str, filepath = "./mlmd", execution_id: str = ""):
+def execution_list(pipeline_name: str, filepath = "./mlmd", execution_uuid: str = ""):
     """Displays executions from the MLMD file with a few properties in a 7-column table, limited to 20 records per page.
     Example: 
     ```python 
-        result = _execution_list("example_pipeline", "./mlmd_directory", "example_execution_id") 
+        result = _execution_list("example_pipeline", "./mlmd_directory", "example_execution_uuid") 
     ```
     Args: 
        pipeline_name: Name of the pipeline. 
        filepath: Path to store the mlmd file. 
-       execution_id: Executions for particular execution id.
+       execution_uuid: Executions for particular execution uuid.
     Returns:
        Output from the _execution_list function. 
     """
 
     # Required arguments: pipeline_name
-    # Optional arguments: filepath( path to store mlmd file), execution_id
-    output = _execution_list(pipeline_name, filepath, execution_id)
+    # Optional arguments: filepath( path to store mlmd file), execution_uuid
+    output = _execution_list(pipeline_name, filepath, execution_uuid)
     return output
 
 
@@ -2369,3 +2371,44 @@ def artifact_list(pipeline_name: str, filepath = "./mlmd", artifact_name: str =
     # Optional arguments: filepath( path to store mlmd file), artifact_name
     output = _artifact_list(pipeline_name, filepath, artifact_name)
     return output
+
+
+def repo_push(pipeline_name: str, filepath = "./mlmd", tensorboard_path: str = "", execution_uuid: str = ""):
+    """ Push artifacts, metadata files, and source code to the user's artifact repository, cmf-server, and git respectively.
+    Example: 
+    ```python 
+        result = _repo_push("example_pipeline", "./mlmd_directory", "example_execution_uuid", "./tensorboard_path") 
+    ```
+    Args: 
+       pipeline_name: Name of the pipeline. 
+       filepath: Path to store the mlmd file.
+       execution_uuid: Executions for particular execution uuid.
+       tensorboard_path: Path to tensorboard logs.
+    Returns:
+       Output from the _repo_push function. 
+    """
+
+    # Required arguments: pipeline_name
+    # Optional arguments: filepath, execution_uuid, tensorboard_path
+    output = _repo_push(pipeline_name, filepath, execution_uuid, tensorboard_path)
+    return output
+
+
+def repo_pull(pipeline_name: str, filepath = "./mlmd", execution_uuid: str = ""):
+    """ Pull artifacts, metadata files, and source code from the user's artifact repository, cmf-server, and git respectively.
+    Example: 
+    ```python 
+        result = _repo_pull("example_pipeline", "./mlmd_directory", "example_execution_uuid") 
+    ```
+    Args: 
+       pipeline_name: Name of the pipeline. 
+       filepath: Path to store the mlmd file. 
+       execution_uuid: Executions for particular execution uuid.
+    Returns:
+       Output from the _repo_pull function. 
+    """
+
+    # Required arguments: pipeline_name
+    # Optional arguments: filepath, execution_uuid
+    output = _repo_pull(pipeline_name, filepath, execution_uuid)
+    return output

cmflib/cmf_commands_wrapper.py

@@ -17,7 +17,7 @@
 from cmflib import cli
 
 
-def _metadata_push(pipeline_name, file_name, execution_id, tensorboard):
+def _metadata_push(pipeline_name, file_name, execution_uuid, tensorboard):
     cli_args = cli.parse_args(
             [
                "metadata",
@@ -27,7 +27,7 @@ def _metadata_push(pipeline_name, file_name, execution_id, tensorboard):
                "-f",
                file_name,
                "-e",
-               execution_id,
+               execution_uuid,
                "-t",
                tensorboard
             ]
@@ -37,7 +37,7 @@ def _metadata_push(pipeline_name, file_name, execution_id, tensorboard):
     print(msg)
     return msg
 
-def _metadata_pull(pipeline_name, file_name, execution_id):
+def _metadata_pull(pipeline_name, file_name, execution_uuid):
     cli_args = cli.parse_args(
             [
                "metadata",
@@ -47,7 +47,7 @@ def _metadata_pull(pipeline_name, file_name, execution_id):
                "-f",
                file_name,
                "-e",
-               execution_id,
+               execution_uuid,
             ]
            )
     cmd = cli_args.func(cli_args)
@@ -316,7 +316,7 @@ def _pipeline_list(file_name):
     print(msg)
     return msg
 
-def _execution_list(pipeline_name, file_name, execution_id):
+def _execution_list(pipeline_name, file_name, execution_uuid):
     cli_args = cli.parse_args(
             [
                "execution",
@@ -326,7 +326,45 @@ def _execution_list(pipeline_name, file_name, execution_id):
                "-f",
                file_name,
                "-e",
-               execution_id
+               execution_uuid
+            ]
+           )
+    cmd = cli_args.func(cli_args)
+    msg = cmd.do_run()
+    print(msg)
+    return msg
+
+def _repo_push(pipeline_name, file_name, tensorboard_path, execution_uuid):
+    cli_args = cli.parse_args(
+            [
+               "repo",
+               "push",
+               "-p",
+               pipeline_name,
+               "-f",
+               file_name,
+               "-e",
+               execution_uuid,
+               "-t",
+               tensorboard_path
+            ]
+           )
+    cmd = cli_args.func(cli_args)
+    msg = cmd.do_run()
+    print(msg)
+    return msg
+
+def _repo_pull(pipeline_name, file_name, execution_uuid):
+    cli_args = cli.parse_args(
+            [
+               "repo",
+               "pull",
+               "-p",
+               pipeline_name,
+               "-f",
+               file_name,
+               "-e",
+               execution_uuid
             ]
            )
     cmd = cli_args.func(cli_args)

cmflib/cmf_exception_handling.py

@@ -201,13 +201,13 @@ def handle(self):
         return f"ERROR: Executions not found."
 
 
-class ExecutionIDNotFound(CmfFailure):
-    def __init__(self, exec_id, return_code=105):
-        self.exec_id = exec_id
+class ExecutionUUIDNotFound(CmfFailure):
+    def __init__(self, exec_uuid, return_code=105):
+        self.exec_uuid = exec_uuid
         super().__init__(return_code)
 
     def handle(self):
-        return f"ERROR: Execution id {self.exec_id} is not present in mlmd."
+        return f"ERROR: Execution uuid {self.exec_uuid} is not present in mlmd."
 
 
 class ArtifactNotFound(CmfFailure):

cmflib/cmf_merger.py

@@ -23,7 +23,7 @@
 from ml_metadata.proto import metadata_store_pb2 as mlpb
 from typing import Union
 
-def parse_json_to_mlmd(mlmd_json, path_to_store: str, cmd: str, exec_id: Union[str, int]) -> Union[str, None]:
+def parse_json_to_mlmd(mlmd_json, path_to_store: str, cmd: str, exec_uuid: Union[str, str]) -> Union[str, None]:
     try:
         mlmd_data = json.loads(mlmd_json)
         pipelines = mlmd_data["Pipeline"]
@@ -52,17 +52,16 @@ def parse_json_to_mlmd(mlmd_json, path_to_store: str, cmd: str, exec_id: Union[s
                             graph=graph, is_server=True)
 
         for stage in data["Pipeline"][0]["stages"]:  # Iterates over all the stages
-            if exec_id is None:  #if exec_id is None we pass all the executions.
+            if exec_uuid is None:  #if exec_uuid is None we pass all the executions.
                 list_executions = [execution for execution in stage["executions"]]
-            elif exec_id is not None:  # elif exec_id is not None, we pass executions for that specific id.
+            elif exec_uuid is not None:  # elif exec_uuid is not None, we pass executions for that specific uuid.
                 list_executions = [
                     execution
                     for execution in stage["executions"]
-                    if execution["id"] == int(exec_id)
+                    if exec_uuid in execution['properties']["Execution_uuid"].split(",") 
                 ]
             else:
-                return "Invalid execution id given."
-
+                return "Invalid execution uuid given."
             for execution in list_executions:  # Iterates over all the executions
                 try:
                     _ = cmf_class.merge_created_context(

cmflib/cmfquery.py

@@ -244,7 +244,7 @@ def _get_executions(self, stage_id: int, execution_id: t.Optional[int] = None) -
 
         Args:
             stage_id: Stage identifier.
-            execution_id: If not None, return only execution with this ID.
+            execution_id: If not None, return execution with this ID.
         Returns:
             List of executions matching input parameters.
         """
@@ -889,17 +889,14 @@ def get_one_hop_parent_artifacts_with_id(self, artifact_id: int) -> pd.DataFrame
                 )
         return df
 
-    def dumptojson(self, pipeline_name: str, exec_id: t.Optional[int] = None) -> t.Optional[str]:
+    def dumptojson(self, pipeline_name: str, exec_uuid: t.Optional[str] = None) -> t.Optional[str]:
         """Return JSON-parsable string containing details about the given pipeline.
         Args:
             pipeline_name: Name of an AI pipelines.
-            exec_id: Optional stage execution ID - filter stages by this execution ID.
+            exec_uuid: Optional stage execution_uuid - filter stages by this execution_uuid.
         Returns:
             Pipeline in JSON format.
         """
-        if exec_id is not None:
-            exec_id = int(exec_id)
-
         def _get_node_attributes(_node: t.Union[mlpb.Context, mlpb.Execution, mlpb.Event], _attrs: t.Dict) -> t.Dict:
             for attr in CONTEXT_LIST:
                 #Artifacts getattr call on Type was giving empty string, which was overwriting 
@@ -921,7 +918,7 @@ def _get_node_attributes(_node: t.Union[mlpb.Context, mlpb.Execution, mlpb.Event
             pipeline_attrs = _get_node_attributes(pipeline, {"stages": []})
             for stage in self._get_stages(pipeline.id):
                 stage_attrs = _get_node_attributes(stage, {"executions": []})
-                for execution in self._get_executions(stage.id, execution_id=exec_id):
+                for execution in self.get_all_executions_by_uuid(stage.id, execution_uuid=exec_uuid):
                     # name will be an empty string for executions that are created with
                     # create new execution as true(default)
                     # In other words name property will there only for execution
@@ -982,6 +979,24 @@ def get_all_executions_for_artifact_id(self, artifact_id: int) -> pd.DataFrame:
         except:
             return df
         return df
+
+    def get_all_executions_by_uuid(self, stage_id: int, execution_uuid: t.Optional[str] = None) -> t.List[mlpb.Execution]:
+        """Return executions of the given stage.
+        Args:
+            stage_id: Stage identifier.
+            execution_uuid: If not None, return execution with this uuid.
+        Returns:
+            List of executions matching input parameters.
+        """
+        executions: t.List[mlpb.Execution] = self.store.get_executions_by_context(stage_id)
+        if execution_uuid is None:
+            return executions
+        executions_with_uuid = []
+        for execution in executions:
+            exec_uuid_list = execution.properties['Execution_uuid'].string_value.split(",")
+            if execution_uuid in exec_uuid_list:
+                executions_with_uuid.append(execution)
+        return executions_with_uuid
 
     """def materialize(self, artifact_name:str):
        artifacts = self.store.get_artifacts()