comments

integration_tests/sdk/aqueduct_tests/flow_test.py

@@ -529,22 +529,27 @@ def check_enabled(count):
     assert (
         latest_run.artifact("extract artifact").execution_state().status == ExecutionStatus.DELETED
     )
+    assert latest_run.artifact("extract artifact").get() is None
     assert (
         latest_run.artifact("op_disabled_by_wf artifact").execution_state().status
         == ExecutionStatus.DELETED
     )
+    assert latest_run.artifact("op_disabled_by_wf artifact").get() is None
     assert (
         latest_run.artifact("metric_enabled artifact").execution_state().status
         == ExecutionStatus.SUCCEEDED
     )
+    assert latest_run.artifact("metric_enabled artifact").get() == 100
     assert (
         latest_run.artifact("metric_disabled artifact").execution_state().status
         == ExecutionStatus.DELETED
     )
+    assert latest_run.artifact("metric_disabled artifact").get() is None
     assert (
         latest_run.artifact("check_enabled artifact").execution_state().status
         == ExecutionStatus.SUCCEEDED
     )
+    assert latest_run.artifact("check_enabled artifact").get()
 
 
 def test_artifact_set_name(client, flow_name, engine):

sdk/aqueduct/artifacts/base_artifact.py

@@ -6,7 +6,7 @@
 import numpy as np
 from aqueduct.constants.enums import ArtifactType, OperatorType
 from aqueduct.models.dag import DAG
-from aqueduct.models.execution_state import ExecutionState
+from aqueduct.models.execution_state import ExecutionState, ExecutionStatus
 from aqueduct.type_annotations import Number
 from aqueduct.utils.naming import sanitize_artifact_name
 
@@ -38,6 +38,13 @@ def __init__(
         # It stays 'None' when the artifact runs in previews.
         self._execution_state = execution_state
 
+    def _is_content_deleted(self) -> bool:
+        if self._from_flow_run:
+            if self._execution_state and self._execution_state.status == ExecutionStatus.DELETED:
+                return True
+
+        return False
+
     def id(self) -> uuid.UUID:
         """Fetch the id associated with this artifact.
 

sdk/aqueduct/artifacts/bool_artifact.py

@@ -48,7 +48,7 @@ def __init__(
     ):
         super().__init__(dag, artifact_id, content, from_flow_run, execution_state)
 
-    def get(self, parameters: Optional[Dict[str, Any]] = None) -> Union[bool, np.bool_]:
+    def get(self, parameters: Optional[Dict[str, Any]] = None) -> Optional[Union[bool, np.bool_]]:
         """Materializes a BoolArtifact into a boolean.
 
         Returns:
@@ -60,6 +60,9 @@ def get(self, parameters: Optional[Dict[str, Any]] = None) -> Union[bool, np.boo
             InternalServerError:
                 An unexpected error occurred in the server.
         """
+        if self._is_content_deleted():
+            return None
+
         self._dag.must_get_artifact(self._artifact_id)
 
         if self._from_flow_run:

sdk/aqueduct/artifacts/generic_artifact.py

@@ -51,6 +51,9 @@ def get(self, parameters: Optional[Dict[str, Any]] = None) -> Any:
             InternalServerError:
                 An unexpected error occurred in the server.
         """
+        if self._is_content_deleted():
+            return None
+
         self._dag.must_get_artifact(self._artifact_id)
 
         if self._from_flow_run:

sdk/aqueduct/artifacts/numeric_artifact.py

@@ -73,7 +73,7 @@ def __init__(
     ):
         super().__init__(dag, artifact_id, content, from_flow_run, execution_state)
 
-    def get(self, parameters: Optional[Dict[str, Any]] = None) -> Number:
+    def get(self, parameters: Optional[Dict[str, Any]] = None) -> Optional[Number]:
         """Materializes a NumericArtifact into its immediate float value.
 
         Returns:
@@ -86,6 +86,8 @@ def get(self, parameters: Optional[Dict[str, Any]] = None) -> Number:
                 An unexpected error occurred within the Aqueduct cluster.
         """
         self._dag.must_get_artifact(self._artifact_id)
+        if self._is_content_deleted():
+            return None
 
         if self._from_flow_run:
             if self._get_content() is None:

sdk/aqueduct/artifacts/table_artifact.py

@@ -67,7 +67,7 @@ def __init__(
     ):
         super().__init__(dag, artifact_id, content, from_flow_run, execution_state)
 
-    def get(self, parameters: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
+    def get(self, parameters: Optional[Dict[str, Any]] = None) -> Optional[pd.DataFrame]:
         """Materializes TableArtifact into an actual dataframe.
 
         Args:
@@ -85,6 +85,8 @@ def get(self, parameters: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
                 An unexpected error occurred within the Aqueduct cluster.
         """
         self._dag.must_get_artifact(self._artifact_id)
+        if self._is_content_deleted():
+            return None
 
         if self._from_flow_run:
             if self._get_content() is None:
@@ -110,7 +112,9 @@ def get(self, parameters: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
         assert isinstance(content, pd.DataFrame)
         return content
 
-    def head(self, n: int = 5, parameters: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
+    def head(
+        self, n: int = 5, parameters: Optional[Dict[str, Any]] = None
+    ) -> Optional[pd.DataFrame]:
         """Returns a preview of the table artifact.
 
         >>> db = client.resource(name="demo/")
@@ -125,6 +129,9 @@ def head(self, n: int = 5, parameters: Optional[Dict[str, Any]] = None) -> pd.Da
             A dataframe containing the table contents of this artifact.
         """
         df = self.get(parameters=parameters)
+        if df is None:
+            return None
+
         return df.head(n)
 
     PRESET_METRIC_LIST = ["number_of_missing_values", "number_of_rows", "max", "min", "mean", "std"]