Added support for FailedData to Dataset

src/omnipy/api/exceptions.py

@@ -15,3 +15,7 @@ class OmnipyNoneIsNotAllowedError(NoneIsNotAllowedError):
 
 class PendingDataError(Exception):
     ...
+
+
+class FailedDataError(Exception):
+    ...

src/omnipy/data/helpers.py

@@ -211,3 +211,9 @@ def obj_or_model_contents_isinstance(__obj: object, __class_or_tuple: type) -> b
 @dataclass(frozen=True)
 class PendingData:
     job_name: str
+
+
+@dataclass(frozen=True)
+class FailedData:
+    job_name: str
+    exception: Exception

src/omnipy/data/mixins/pending.py

@@ -2,10 +2,10 @@
 
 from typing_extensions import TypeVar
 
-from omnipy.api.exceptions import PendingDataError
+from omnipy.api.exceptions import FailedDataError, PendingDataError
 from omnipy.api.protocols.public.data import IsDataset
 from omnipy.api.typedefs import TypeForm
-from omnipy.data.helpers import is_model_subclass, PendingData
+from omnipy.data.helpers import FailedData, is_model_subclass, PendingData
 from omnipy.util.decorators import call_super_if_available
 from omnipy.util.helpers import is_union
 
@@ -20,15 +20,15 @@ class PendingDatasetMixin:
     def _prepare_params(cls, params: TypeForm) -> TypeForm:
         cleaned_params = cls._clean_model(params)
         if is_model_subclass(cleaned_params):
-            return cleaned_params | PendingData
+            return cleaned_params | PendingData | FailedData
         else:
             return cleaned_params
 
     @call_super_if_available(call_super_before_method=True)
     @classmethod
     def _clean_model(cls, model: TypeForm) -> TypeForm:
         args = get_args(model)
-        if is_union(model) and len(args) == 2 and args[-1] is PendingData:
+        if is_union(model) and len(args) == 3 and args[1:] == (PendingData, FailedData):
             return args[0]
         else:
             return model
@@ -42,12 +42,24 @@ def pending_data(self) -> IsDataset[type[ModelT]]:
     @property
     def available_data(self) -> IsDataset[type[ModelT]]:
         copy = self.__class__()
-        copy.data = {key: val for key, val in self.data.items() if not isinstance(val, PendingData)}
+        copy.data = {
+            key: val for key,
+            val in self.data.items() if not isinstance(val, (PendingData, FailedData))
+        }
+        return copy
+
+    @property
+    def failed_data(self) -> IsDataset[type[ModelT]]:
+        copy = self.__class__()
+        copy.data = {key: val for key, val in self.data.items() if isinstance(val, FailedData)}
         return copy
 
     @call_super_if_available(call_super_before_method=True)
     def _check_value(self, value: Any) -> Any:
         if isinstance(value, PendingData):
-            raise PendingDataError('Data is still pending')
+            raise PendingDataError(f'Dataset is still awaiting data from job "{value.job_name}"')
+        elif isinstance(value, FailedData):
+            raise FailedDataError(
+                f'Job "{value.job_name}" failed to return data: {repr(value.exception)}')
 
         return value

tests/data/test_dataset.py

@@ -8,10 +8,10 @@
 import pytest_cases as pc
 from typing_extensions import TypeVar
 
-from omnipy.api.exceptions import PendingDataError
+from omnipy.api.exceptions import FailedDataError, PendingDataError
 from omnipy.api.protocols.public.hub import IsRuntime
 from omnipy.data.dataset import Dataset
-from omnipy.data.helpers import PendingData
+from omnipy.data.helpers import FailedData, PendingData
 from omnipy.data.model import Model
 
 from .helpers.classes import MyFloatObject
@@ -1191,61 +1191,118 @@ def test_dataset_switch_models_issue():
     dataset['a'], dataset['b'] = dataset['b'], dataset['a']
 
 
-def _assert_pending_data_error(dataset: Dataset):
-    with pytest.raises(PendingDataError):
+def _assert_no_access_data_exceptions(dataset: Dataset,
+                                      model_cls: type,
+                                      keys: list[str],
+                                      values: list) -> None:
+    assert len(keys) == len(values) == len(dataset)
+    model_values = [model_cls(value) for value in values]
+    assert [_ for _ in dataset.values()] == model_values
+    assert [_ for _ in dataset.items()] == list(zip(keys, model_values))
+    for i, key in enumerate(keys):
+        assert dataset[key] == model_values[i]
+    assert dict(dataset) == dict(zip(keys, model_values))
+    assert Dataset[Model[str]](dataset) == dataset
+
+
+def _assert_access_data_exception(dataset: Dataset, exception_cls: type[Exception],
+                                  keys: list[str]) -> None:
+    with pytest.raises(exception_cls):
         [_ for _ in dataset.values()]
 
-    with pytest.raises(PendingDataError):
+    with pytest.raises(exception_cls):
         [_ for _ in dataset.items()]
 
-    with pytest.raises(PendingDataError):
-        dataset['new_data']
+    with pytest.raises(exception_cls):
+        for key in keys:
+            dataset[key]
 
-    with pytest.raises(PendingDataError):
+    with pytest.raises(exception_cls):
         dict(dataset)
 
-    with pytest.raises(PendingDataError):
+    with pytest.raises(exception_cls):
         Dataset[Model[str]](dataset)
 
 
-def test_dataset_pending_data():
+def _assert_dataset(
+    dataset: Dataset,
+    model_cls: type,
+    keys: list[str],
+    error_cls: type[Exception] | None,
+    values: list | None = None,
+) -> None:
+    assert len(dataset) == len(keys)
+    assert isinstance(dataset, Dataset)
+    assert dataset.get_model_class() == model_cls
+    assert list(dataset.keys()) == keys
+    if error_cls is None:
+        assert values is not None
+        _assert_no_access_data_exceptions(dataset, model_cls, keys, values)
+    else:
+        _assert_access_data_exception(dataset, error_cls, keys)
+
+
+def test_dataset_pending_and_failed_data() -> None:
     dataset = Dataset[Model[str]]()
 
-    dataset['old_data'] = 'This is some existing data'
-    dataset['new_data'] = PendingData('my_task')
+    dataset['old_data'] = 'Existing data'
+    _assert_dataset(dataset, Model[str], ['old_data'], None, ['Existing data'])
 
-    assert len(dataset) == 2
-    assert list(dataset.keys()) == ['old_data', 'new_data']
-    _assert_pending_data_error(dataset)
-
-    pending_data = dataset.pending_data
-    assert isinstance(pending_data, Dataset)
-    assert pending_data.get_model_class() == Model[str]
-    assert len(pending_data) == 1
-    assert list(pending_data.keys()) == ['new_data']
-    _assert_pending_data_error(pending_data)
-
-    available_data = dataset.available_data
-    assert len(available_data) == 1
-    assert isinstance(available_data, Dataset)
-    assert available_data.get_model_class() == Model[str]
-    assert len(available_data) == 1
-    assert list(available_data.keys()) == ['old_data']
-
-    assert list(available_data.values()) == [Model[str]('This is some existing data')]
-    assert list(available_data.items()) == [('old_data', Model[str]('This is some existing data'))]
-
-    dataset['new_data'] = 'New data is now available'
-    assert len(dataset.pending_data) == 0
-    assert len(dataset.available_data) == 2
-
-    available_data = dataset.available_data
-    assert list(available_data.keys()) == ['old_data', 'new_data']
-    assert list(available_data.values()) == [
-        Model[str]('This is some existing data'), Model[str]('New data is now available')
-    ]
-    assert list(available_data.items()) == [('old_data', Model[str]('This is some existing data')),
-                                            ('new_data', Model[str]('New data is now available'))]
+    dataset['new_data'] = PendingData('my_task')
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data'], PendingDataError)
+    _assert_dataset(dataset.pending_data, Model[str], ['new_data'], PendingDataError)
+    _assert_dataset(dataset.available_data, Model[str], ['old_data'], None, ['Existing data'])
+    _assert_dataset(dataset.failed_data, Model[str], [], None, [])
+
+    dataset['newer_data'] = PendingData('my_other_task')
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data', 'newer_data'], PendingDataError)
+    _assert_dataset(dataset.pending_data, Model[str], ['new_data', 'newer_data'], PendingDataError)
+    _assert_dataset(dataset.available_data, Model[str], ['old_data'], None, ['Existing data'])
+    _assert_dataset(dataset.failed_data, Model[str], [], None, [])
+
+    dataset['new_data'] = FailedData('my_task', RuntimeError('Something went wrong'))
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data', 'newer_data'], FailedDataError)
+    _assert_dataset(dataset.pending_data, Model[str], ['newer_data'], PendingDataError)
+    _assert_dataset(dataset.available_data, Model[str], ['old_data'], None, ['Existing data'])
+    _assert_dataset(dataset.failed_data, Model[str], ['new_data'], FailedDataError)
+
+    dataset['newer_data'] = FailedData('my_other_task', ValueError('Something else went wrong'))
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data', 'newer_data'], FailedDataError)
+    _assert_dataset(dataset.pending_data, Model[str], [], None, [])
+    _assert_dataset(dataset.available_data, Model[str], ['old_data'], None, ['Existing data'])
+    _assert_dataset(dataset.failed_data, Model[str], ['new_data', 'newer_data'], FailedDataError)
+
+    dataset['newer_data'] = PendingData('my_retry_task')
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data', 'newer_data'], FailedDataError)
+    _assert_dataset(dataset.pending_data, Model[str], ['newer_data'], PendingDataError)
+    _assert_dataset(dataset.available_data, Model[str], ['old_data'], None, ['Existing data'])
+    _assert_dataset(dataset.failed_data, Model[str], ['new_data'], FailedDataError)
+
+    dataset['new_data'] = 'Fixed data'
+    _assert_dataset(dataset, Model[str], ['old_data', 'new_data', 'newer_data'], PendingDataError)
+    _assert_dataset(dataset.pending_data, Model[str], ['newer_data'], PendingDataError)
+    _assert_dataset(dataset.available_data,
+                    Model[str], ['old_data', 'new_data'],
+                    None, ['Existing data', 'Fixed data'])
+    _assert_dataset(dataset.failed_data, Model[str], [], None, [])
+
+    dataset['newer_data'] = 'Retried data'
+    _assert_dataset(
+        dataset,
+        Model[str],
+        ['old_data', 'new_data', 'newer_data'],
+        None,
+        ['Existing data', 'Fixed data', 'Retried data'],
+    )
+    _assert_dataset(dataset.pending_data, Model[str], [], None, [])
+    _assert_dataset(
+        dataset.available_data,
+        Model[str],
+        ['old_data', 'new_data', 'newer_data'],
+        None,
+        ['Existing data', 'Fixed data', 'Retried data'],
+    )
+    _assert_dataset(dataset.failed_data, Model[str], [], None, [])
 
 
 # TODO: Add unit tests for MultiModelDataset

tests/data/test_helpers.py

@@ -3,7 +3,10 @@
 from pydantic import ValidationError
 import pytest
 
-from omnipy.data.helpers import is_model_instance, obj_or_model_contents_isinstance, PendingData
+from omnipy.data.helpers import (FailedData,
+                                 is_model_instance,
+                                 obj_or_model_contents_isinstance,
+                                 PendingData)
 from omnipy.data.model import Model
 
 from .helpers.models import PydanticParentModel
@@ -73,7 +76,24 @@ def test_pending_data() -> None:
     assert pending_data.job_name == 'my_task'
 
     with pytest.raises(AttributeError):
-        pending_data.job_name = 'my_other_task'  # type: ignore[misc]
+        pending_data.job_name = 'my_other_task'
 
     with pytest.raises(ValidationError):
         Model[str](pending_data)
+
+
+def test_failed_data() -> None:
+    with pytest.raises(TypeError):
+        FailedData()  # type: ignore[call-arg]
+
+    exception = RuntimeError('Some error')
+    error_data = FailedData('my_task', exception)
+    assert error_data.job_name == 'my_task'
+    assert error_data.exception is exception
+
+    with pytest.raises(AttributeError):
+        error_data.job_name = 'my_other_task'
+        error_data.exception = Exception('other errors')
+
+    with pytest.raises(ValidationError):
+        Model[str](error_data)