exposes current pipeline code, type

Author: nazarfil

openhexa/sdk/__init__.py

@@ -1,7 +1,7 @@
 """SDK package."""
 
 from .datasets import Dataset
-from .pipelines import current_run, parameter, pipeline
+from .pipelines import current_pipeline, current_run, parameter, pipeline
 from .pipelines.parameter import DHIS2Widget, IASOWidget
 from .utils import OpenHexaClient
 from .workspaces import workspace
@@ -17,6 +17,7 @@
 __all__ = [
     "workspace",
     "pipeline",
+    "current_pipeline",
     "parameter",
     "current_run",
     "DHIS2Connection",

openhexa/sdk/pipelines/__init__.py

@@ -15,6 +15,7 @@
     "Pipeline",
     "parameter",
     "current_run",
+    "current_pipeline",
     "import_pipeline",
     "download_pipeline",
     "get_local_workspace_config",

openhexa/sdk/pipelines/current_pipeline.py

@@ -0,0 +1,47 @@
+"""Current Pipeline context module."""
+import os
+
+from openhexa.sdk.utils import Environment, get_environment
+
+
+class PipelineConfigError(Exception):
+    """Raised whenever the system cannot find an environment variable required to configure the current pipeline."""
+
+    pass
+
+
+class CurrentPipeline:
+    """It is used to view manage the current pipeline state and operations."""
+
+    @property
+    def code(self) -> str:
+        """The unique slug of the workspace.
+
+        Slugs are used to identify the workspace.
+        """
+        try:
+            return os.environ["HEXA_PIPELINE_CODE"]
+        except KeyError:
+            raise PipelineConfigError("The pipeline code is not available in this environment.")
+
+    @property
+    def name(self) -> str:
+        """The name of the current pipeline."""
+        try:
+            return os.environ["HEXA_PIPELINE_NAME"]
+        except KeyError:
+            raise PipelineConfigError("The pipeline name is not available in this environment.")
+
+    @property
+    def type(self) -> str:
+        """The type of the current pipeline."""
+        try:
+            return os.environ["HEXA_PIPELINE_TYPE"]
+        except KeyError:
+            raise PipelineConfigError("The pipeline type is not available in this environment.")
+
+
+if get_environment() == Environment.CLOUD_JUPYTER:
+    current_pipeline = None
+else:
+    current_pipeline = CurrentPipeline()