"""This module contains the mosaic-orchestrator framework core functionality"""
from typing import (
Any,
Protocol,
runtime_checkable,
)
from mosaic_orchestrator.result import (
Validation,
)
from typing import Protocol
[docs]@runtime_checkable
class Hashable(Protocol):
"""Protocol that is used during caching for task state representation.
When a member variable of a task complies with this protocol the `task_hash` method is called
to represent its state.
"""
# pylint: disable=too-few-public-methods
[docs] def task_hash(self) -> Any:
"""Returns a hash that represents this objects state.
Returns:
Anything that represents this objects state. It must be pickle-able, i.e. compatible
with the pickle.dumps() function of the standard library.
"""
pass
[docs]@runtime_checkable
class LifeCycleListener(Protocol):
"""The methods of this protocol are called before startup and after teardown of a run."""
# pylint: disable=too-few-public-methods
def _on_start(self):
"""called before the run of the root task."""
def _on_end(self, exception: Exception = None):
"""called after the run of the root task.
This method is called even if an exception is thrown during the run, similar to
try-catch-finally in python.
Args:
exception: if the run aborted with an exception it is passed as an argument.
"""
[docs]@runtime_checkable
class Validate(Protocol):
"""Protocol that is used to indentify and validate task members.
When a member variable of a task object conforms to this protocol the `validate()` function
will be called during initialization of the task hierarchy. In case validation is unsuccessful
an error is reported.
"""
# pylint: disable=too-few-public-methods
[docs] def validate(self) -> Validation:
"""validates this object during initialization.
This function is called before any task is executed.
Returns:
`ValidationSuccess` or `ValidationError` in case of an error
"""
pass