2.8.0 (#72)

Author: z7ye

ads/ads_version.json

@@ -1,3 +1,3 @@
 {
-  "version": "2.7.3"
+  "version": "2.8.0"
 }

ads/cli.py

@@ -7,13 +7,16 @@
     import click
     import ads.opctl.cli
     import ads.jobs.cli
+    import ads.pipeline.cli
     import os
     import json
 except:
     print(
         "Please run `pip install oracle-ads[opctl]` to install the required dependencies for ADS CLI"
     )
     exit()
+
+
 with open(
     os.path.join(os.path.dirname(os.path.abspath(__file__)), "ads_version.json")
 ) as version_file:
@@ -29,6 +32,7 @@ def cli():
 
 cli.add_command(ads.opctl.cli.commands)
 cli.add_command(ads.jobs.cli.commands)
+cli.add_command(ads.pipeline.cli.commands)
 
 
 if __name__ == "__main__":

ads/common/auth.py

@@ -7,18 +7,15 @@
 import copy
 import os
 from dataclasses import dataclass
-from typing import Callable, Dict, Optional, Any
-
-import oci
-from oci.config import (
-    DEFAULT_LOCATION,  # "~/.oci/config"
-    DEFAULT_PROFILE,  # "DEFAULT"
-)
+from typing import Any, Callable, Dict, Optional
 
 import ads.telemetry
+import oci
 from ads.common import logger
 from ads.common.decorator.deprecate import deprecated
 from ads.common.extended_enum import ExtendedEnumMeta
+from oci.config import DEFAULT_LOCATION  # "~/.oci/config"
+from oci.config import DEFAULT_PROFILE  # "DEFAULT"
 
 
 class AuthType(str, metaclass=ExtendedEnumMeta):
@@ -314,7 +311,9 @@ def create_signer(
         )
         if config:
             auth_type = AuthType.API_KEY
+
         signer_generator = AuthFactory().signerGenerator(auth_type)
+
         return signer_generator(signer_args).create_signer()
 
 

ads/jobs/ads_job.py

@@ -49,7 +49,7 @@ class Job(Builder):
             .with_runtime(
                 ScriptRuntime()
                 .with_source("oci://bucket_name@namespace/path/to/script.py")
-                .with_service_conda("tensorflow28_p38_cpu_v1")
+                .with_service_conda("tensorflow26_p37_cpu_v2")
                 .with_environment_variable(ENV="value")
                 .with_argument("argument", key="value")
                 .with_freeform_tag(tag_name="tag_value")
@@ -79,7 +79,7 @@ class Job(Builder):
             .with_runtime(
                 NotebookRuntime()
                 .with_notebook("path/to/notebook.ipynb")
-                .with_service_conda("tensorflow28_p38_cpu_v1")
+                .with_service_conda(tensorflow26_p37_cpu_v2")
                 # Saves the notebook with outputs to OCI object storage.
                 .with_output("oci://bucket_name@namespace/path/to/dir")
             )

ads/model/model_metadata.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*--
 
-# Copyright (c) 2021, 2022 Oracle and/or its affiliates.
+# Copyright (c) 2021, 2023 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
 import json
@@ -38,6 +38,7 @@
 METADATA_VALUE_LENGTH_LIMIT = 255
 METADATA_DESCRIPTION_LENGTH_LIMIT = 255
 _METADATA_EMPTY_VALUE = "NA"
+CURRENT_WORKING_DIR = "."
 
 
 class MetadataSizeTooLarge(ValueError):
@@ -1594,7 +1595,16 @@ def fetch_training_code_details(
         ModelProvenanceMetadata
             A ModelProvenanceMetadata instance.
         """
-        repo = git.Repo(".", search_parent_directories=True)
+        git_dir = CURRENT_WORKING_DIR
+        if training_script_path:
+            if not os.path.exists(training_script_path):
+                logger.warning(
+                    f"Training script {os.path.abspath(training_script_path)} does not exist."
+                )
+            else:
+                training_script_path = os.path.abspath(training_script_path)
+                git_dir = os.path.dirname(training_script_path)
+        repo = git.Repo(git_dir, search_parent_directories=True)
         # get repository url
         if len(repo.remotes) > 0:
             repository_url = (
@@ -1608,19 +1618,11 @@ def fetch_training_code_details(
         # get git branch
         git_branch = format(repo.active_branch)
         # get git commit
-        git_commit = ""
+        git_commit = None
         try:
-            git_commit = format(str(repo.head.commit.hexsha))
-        except ValueError:
-            # do not set commit if there isn't any
+            git_commit = format(str(repo.head.commit.hexsha)) or None
+        except Exception:
             logger.warning("No commit found.")
-        if training_script_path is not None and training_script_path != "":
-            if not os.path.exists(training_script_path):
-                logger.warning(
-                    f"Training script {os.path.abspath(training_script_path)} does not exists."
-                )
-            else:
-                training_script_path = os.path.abspath(training_script_path)
         return cls(
             repo=repo,
             git_branch=git_branch,

ads/opctl/backend/ads_dataflow.py

@@ -46,14 +46,22 @@ def __init__(self, config: Dict) -> None:
         self.profile = config["execution"].get("oci_profile", None)
         self.client = OCIClientFactory(**self.oci_auth).dataflow
 
+    def apply(self):
+        """
+        Create DataFlow and DataFlow Run from YAML.
+        """
+        # TODO add the logic for build dataflow and dataflow run from YAML.
+        raise NotImplementedError(f"`apply` hasn't been supported for data flow yet.")
+
     def run(self) -> None:
+        """
+        Create DataFlow and DataFlow Run from OCID or cli parameters.
+        """
         with AuthContext():
             ads.set_auth(auth=self.auth_type, profile=self.profile)
-            if self.config["execution"].get("job_id", None):
-                job_id = self.config["execution"]["job_id"]
-                run_id = (
-                    Job.from_dataflow_job(self.config["execution"]["job_id"]).run().id
-                )
+            if self.config["execution"].get("ocid", None):
+                data_flow_id = self.config["execution"]["ocid"]
+                run_id = Job.from_dataflow_job(data_flow_id).run().id
             else:
                 infra = self.config.get("infrastructure", {})
                 if any(k not in infra for k in REQUIRED_FIELDS):
@@ -85,6 +93,9 @@ def run(self) -> None:
         return {"job_id": job_id, "run_id": run_id}
 
     def cancel(self):
+        """
+        Cancel DataFlow Run from OCID.
+        """
         if not self.config["execution"].get("run_id"):
             raise ValueError("Can only cancel a DataFlow run.")
         run_id = self.config["execution"]["run_id"]
@@ -93,18 +104,24 @@ def cancel(self):
             DataFlowRun.from_ocid(run_id).delete()
 
     def delete(self):
-        if self.config["execution"].get("job_id"):
-            job_id = self.config["execution"]["job_id"]
+        """
+        Delete DataFlow or DataFlow Run from OCID.
+        """
+        if self.config["execution"].get("id"):
+            data_flow_id = self.config["execution"]["id"]
             with AuthContext():
                 ads.set_auth(auth=self.auth_type, profile=self.profile)
-                Job.from_dataflow_job(job_id).delete()
+                Job.from_dataflow_job(data_flow_id).delete()
         elif self.config["execution"].get("run_id"):
             run_id = self.config["execution"]["run_id"]
             with AuthContext():
                 ads.set_auth(auth=self.auth_type, profile=self.profile)
                 DataFlowRun.from_ocid(run_id).delete()
 
     def watch(self):
+        """
+        Watch DataFlow Run from OCID.
+        """
         run_id = self.config["execution"]["run_id"]
         with AuthContext():
             ads.set_auth(auth=self.auth_type, profile=self.profile)

ads/opctl/backend/ads_ml_job.py

@@ -60,16 +60,29 @@ def __init__(self, config: Dict) -> None:
         self.profile = config["execution"].get("oci_profile", None)
         self.client = OCIClientFactory(**self.oci_auth).data_science
 
+    def apply(self) -> None:
+        """
+        Create Job and Job Run from YAML.
+        """
+        with AuthContext():
+            ads.set_auth(auth=self.auth_type, profile=self.profile)
+            job = Job.from_dict(self.config)
+            job.create()
+            job_run = job.run()
+            print("JOB OCID:", job.id)
+            print("JOB RUN OCID:", job_run.id)
+
     def run(self) -> None:
+        """
+        Create Job and Job Run from OCID or cli parameters.
+        """
         # TODO Check that this still runs smoothly for distributed
         with AuthContext():
             ads.set_auth(auth=self.auth_type, profile=self.profile)
-            if self.config["execution"].get("job_id", None):
-                job_id = self.config["execution"]["job_id"]
+            if self.config["execution"].get("ocid", None):
+                job_id = self.config["execution"]["ocid"]
                 run_id = (
-                    Job.from_datascience_job(self.config["execution"]["job_id"])
-                    .run()
-                    .id
+                    Job.from_datascience_job(self.config["execution"]["ocid"]).run().id
                 )
             else:
                 payload = self._create_payload()  # create job with infrastructure
@@ -121,24 +134,36 @@ def init_operator(self):
         return operator_folder
 
     def delete(self):
-        if self.config["execution"].get("job_id"):
-            job_id = self.config["execution"]["job_id"]
+        """
+        Delete Job or Job Run from OCID.
+        """
+        if self.config["execution"].get("id"):
+            job_id = self.config["execution"]["id"]
             with AuthContext():
                 ads.set_auth(auth=self.auth_type, profile=self.profile)
                 Job.from_datascience_job(job_id).delete()
+                print(f"Job {job_id} has been deleted.")
         elif self.config["execution"].get("run_id"):
             run_id = self.config["execution"]["run_id"]
             with AuthContext():
                 ads.set_auth(auth=self.auth_type, profile=self.profile)
                 DataScienceJobRun.from_ocid(run_id).delete()
+                print(f"Job run {run_id} has been deleted.")
 
     def cancel(self):
+        """
+        Cancel Job Run from OCID.
+        """
         run_id = self.config["execution"]["run_id"]
         with AuthContext():
             ads.set_auth(auth=self.auth_type, profile=self.profile)
             DataScienceJobRun.from_ocid(run_id).cancel()
+            print(f"Job run {run_id} has been cancelled.")
 
     def watch(self):
+        """
+        Watch Job Run from OCID.
+        """
         run_id = self.config["execution"]["run_id"]
 
         with AuthContext():

ads/opctl/backend/ads_ml_pipeline.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python
+# -*- coding: utf-8; -*-
+
+# Copyright (c) 2022, 2023 Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
+
+from typing import Dict
+import ads
+from ads.common.auth import create_signer, AuthContext
+from ads.common.oci_client import OCIClientFactory
+from ads.opctl.backend.base import Backend
+from ads.pipeline import Pipeline, PipelineRun
+
+
+class PipelineBackend(Backend):
+    def __init__(self, config: Dict) -> None:
+        """
+        Initialize a MLPipeline object given config dictionary.
+
+        Parameters
+        ----------
+        config: dict
+            dictionary of configurations
+        """
+        self.config = config
+        self.oci_auth = create_signer(
+            config["execution"].get("auth"),
+            config["execution"].get("oci_config", None),
+            config["execution"].get("oci_profile", None),
+        )
+        self.auth_type = config["execution"].get("auth")
+        self.profile = config["execution"].get("oci_profile", None)
+        self.client = OCIClientFactory(**self.oci_auth).data_science
+
+    def apply(self) -> None:
+        """
+        Create Pipeline and Pipeline Run from YAML.
+        """
+        with AuthContext():
+            ads.set_auth(auth=self.auth_type, profile=self.profile)
+            pipeline = Pipeline.from_dict(self.config)
+            pipeline.create()
+            pipeline_run = pipeline.run()
+            print("PIPELINE OCID:", pipeline.id)
+            print("PIPELINE RUN OCID:", pipeline_run.id)
+
+    def run(self) -> None:
+        """
+        Create Pipeline and Pipeline Run from OCID.
+        """
+        pipeline_id = self.config["execution"]["ocid"]
+        with AuthContext():
+            ads.set_auth(auth=self.auth_type, profile=self.profile)
+            pipeline = Pipeline.from_ocid(ocid=pipeline_id)
+            pipeline_run = pipeline.run()
+            print("PIPELINE RUN OCID:", pipeline_run.id)
+
+    def delete(self) -> None:
+        """
+        Delete Pipeline or Pipeline Run from OCID.
+        """
+        if self.config["execution"].get("id"):
+            pipeline_id = self.config["execution"]["id"]
+            with AuthContext():
+                ads.set_auth(auth=self.auth_type, profile=self.profile)
+                Pipeline.from_ocid(pipeline_id).delete()
+                print(f"Pipeline {pipeline_id} has been deleted.")
+        elif self.config["execution"].get("run_id"):
+            run_id = self.config["execution"]["run_id"]
+            with AuthContext():
+                ads.set_auth(auth=self.auth_type, profile=self.profile)
+                PipelineRun.from_ocid(run_id).delete()
+                print(f"Pipeline run {run_id} has been deleted.")
+
+    def cancel(self) -> None:
+        """
+        Cancel Pipeline Run from OCID.
+        """
+        run_id = self.config["execution"]["run_id"]
+        with AuthContext():
+            ads.set_auth(auth=self.auth_type, profile=self.profile)
+            PipelineRun.from_ocid(run_id).cancel()
+            print(f"Pipeline run {run_id} has been cancelled.")
+
+    def watch(self) -> None:
+        """
+        Watch Pipeline Run from OCID.
+        """
+        run_id = self.config["execution"]["run_id"]
+        log_type = self.config["execution"]["log_type"]
+        with AuthContext():
+            ads.set_auth(auth=self.auth_type, profile=self.profile)
+            PipelineRun.from_ocid(run_id).watch(log_type=log_type)