"""This module contains the mosaic-orchestrator framework core functionality"""
from __future__ import annotations
import pathlib
from abc import ABC, abstractmethod
from typing import (
Dict,
Union,
Type,
TypeVar,
Generic,
)
from mosaic_orchestrator.config import Config
from mosaic_orchestrator.pdk import PDK
from mosaic_orchestrator.task import CheckResult
T = TypeVar("T")
[docs]class TaskHierarchyBuilder(ABC, Generic[T]):
"""A builder for a task hierarchy with a specific root task type.
Use the `check` method to discover missing/invalid dependencies or task inputs. When all
errors are resolved use `build` to retrieve the instance of the root task.
"""
[docs] @abstractmethod
def register(
self, key: Union[Type, str], value: any, overwrite: bool = False
) -> TaskHierarchyBuilder[T]:
"""registers a dependency based on its type or its path in the hierarchy.
This is the intended way to make tool implementations available for tasks. Subsequent calls
with the same key overwrite the initial value.
Args:
key: `value` will be injected into all tasks that have members annotated with this type.
When key is `str` it is expected to be a path to a task input; in this case
`value` is set to this input.
value: this object will be injected when `key` matches
overwrite: Overwrite tool/object if it has already been registered
Returns:
a self reference to allow method chaining.
"""
raise NotImplementedError("register not implemented")
[docs] @abstractmethod
def with_pdk(self, pdk: PDK) -> TaskHierarchyBuilder[T]:
"""sets the pdk for this hierarchy. Subsequent calls overwrite the inital value.
Args:
pdk: the instantiated pdk object.
Returns:
a self reference to allow method chaining.
"""
raise NotImplementedError("with_pdk not implemented")
[docs] @abstractmethod
def with_run_directory(self, path: pathlib.Path) -> TaskHierarchyBuilder[T]:
"""sets the run directory for this hierarchy.
Args:
path: a path to the target folder. The folder will be created if it does not exist.
Returns:
a self reference to allow method chaining.
"""
raise NotImplementedError("with_run_directory not implemented")
[docs] @abstractmethod
def with_cache_directory(self, path: pathlib.Path) -> TaskHierarchyBuilder[T]:
"""sets the cache directory for this hierarchy.
Args:
path: a path to the target folder. The folder will be created if it does not exist.
Returns:
a self reference to allow method chaining.
"""
raise NotImplementedError("with_cache_directory not implemented")
[docs] @abstractmethod
def build(self) -> T:
"""builds the task hierarchy and returns the instance of the root task.
Use the `check` method to test if build is going to be successful.
Returns:
the root task instance.
Raises:
MosaicException: in case of a missing dependency or an unsuccessful validation.
"""
raise NotImplementedError("build not implemented")
[docs] @abstractmethod
def check(self) -> CheckResult:
"""checks dependencies and task inputs and performs validation.
This method can be called repeatedly to find remaining errors.
Returns:
A `CheckResult` object containing a list of all settable inputs as well as a list
of remaining errors.
"""
raise NotImplementedError("check not implemented")
[docs] @abstractmethod
def with_cfg(self, config: Union[Config, pathlib.Path]) -> TaskHierarchyBuilder[T]:
"""loads a `Config` either from an object or from a python script.
This method can be called repeatedly to merge multiple configurations.
Args:
config: a `Config` object that will be applied to this hierarchy. Alternatively, pass a
path to a python script containing a `MosaicConfig` class; This class will be
automatically isntantiated and loaded.
Returns:
a self reference to allow method chaining.
"""
raise NotImplementedError("with_cfg not implemented")