fairtracks / omnipy Goto Github PK

        del cls.__annotations__[ROOT_KEY]

    def __class_getitem__(cls, model: Union[Type[RootT], TypeVar]) -> Union[Type[RootT], TypeVar]:
        # TODO: change model type to params: Union[Type[Any], Tuple[Type[Any], ...]]
        #       as in GenericModel

        # For now, only singular model types are allowed. These lines are needed for

            else:
                raise RuntimeError('Model does not allow setting of extra attributes')

    # TODO: Update Dataset.__eq__ similarly, with tests
    def __eq__(self, other: object) -> bool:
        return self.__class__ == other.__class__ and super().__eq__(other)

            result = None
            with linear_flow.flow_context:
                for i, job in enumerate(linear_flow.task_templates):
                    # TODO: Better handling of kwargs
                    if i == 0:
                        result = job(*args, **kwargs)
                    else:

    data: Dict[str, ModelT] = Field(default={})

    def __class_getitem__(cls, model: ModelT) -> ModelT:
        # TODO: change model type to params: Union[Type[Any], Tuple[Type[Any], ...]]
        #       as in GenericModel

        # For now, only singular model types are allowed. These lines are needed for

DagFlow.accept_mixin(FlowContextJobMixin)
FuncFlow.accept_mixin(FlowContextJobMixin)

# TODO: Recursive replace - *args: Any -> *args: object, *kwargs: Any -> *kwargs: object

@TaskTemplate()
def remove_columns(json_dataset: JsonListOfDictsOfAnyDataset,
                   column_keys_for_data_items: Dict[str, List[str]]) -> JsonListOfDictsOfAnyDataset:
    # TODO: implement general solution to make sure that one does not modify input data by
    #       automatically copying or otherwise. Perhaps setting immutable/frozen option in pydantic,
    #       if available?
    #

        return _prefect_flow(*args, **kwargs)


# TODO: Refactor redundant code

    FINISHED = 3


# TODO: Add 'apply' state
# TODO: Add 'failed' state and error management
# TODO: Consider the need for a 'waiting' state


class RunStateLogMessages(str, Enum):

        return multi_model_dataset


# TODO: Use json serializer package from the pydantic config instead of 'json'


class MultiModelDataset(Dataset[ModelT], Generic[ModelT]):

                                        JsonModel,
                                        JsonNestedDictsModel)

# TODO: switch from plural to singular for names of modules in omnipy modules
# TODO: call omnipy modules something else than modules, to distinguish from Python modules.
#       Perhaps plugins?

JsonModelT = TypeVar('JsonModelT', bound=Union[JsonModel, JsonListModel, JsonDictModel])

        else:
            job_obj = cls.__new__(cls, *args, **kwargs)

        # TODO: refactor using state machine
        job_obj._from_apply = True
        job_obj.__init__(*args, **kwargs)
        job_obj._from_apply = False

        datetime_str = run_time.strftime('%Y_%m_%d-%H_%M_%S')
        return datetime_str

    # TODO: Refactor
    def _deserialize_and_restore_outputs(self) -> Dataset:
        output_path = Path(self.config.persist_data_dir_path)
        if os.path.exists(output_path):

        # As long as models are not created concurrently, setting the class members temporarily
        # should not have averse effects
        # TODO: Check if we can move to explicit definition of __root__ field at the object
        #       level in pydantic 2.0 (when it is released)
        if cls == Model:
            cls._depopulate_root_field()

                    param_keys = set(inspect.signature(job).parameters.keys())

                    # TODO: Refactor to remove dependency
                    #       Also, add test for not allowing override of fixed_params
                    if hasattr(job, 'param_key_map'):
                        for key, val in job.param_key_map.items():

            if hasattr(self, '_job_func'):
                self._name = self._job_func.__name__

        # TODO: When job state machine is implemented, check using that to see if in job state
        self._unique_name = self._generate_unique_name() if hasattr(self, 'create_job') else None

    @staticmethod

            return self._persist_outputs if self._persist_outputs is not None \
                    else PersistOpts.DISABLED
        else:
            # TODO: Refactor using Flow and Task Mixins
            from omnipy.compute.flow import FlowBase
            from omnipy.compute.task import TaskBase

                 persist_outputs: Optional[PersistOutputsOptions] = None,
                 restore_outputs: Optional[RestoreOutputsOptions] = None):

        # TODO: Possibly reimplement logic using a state machine, e.g. "transitions" package
        if persist_outputs is None:
            self._persist_outputs = PersistOpts.FOLLOW_CONFIG if self._has_job_config else None
        else:

    FINISHED = 3


# TODO: Add 'apply' state
# TODO: Add 'failed' state and error management
# TODO: Consider the need for a 'waiting' state


class RunStateLogMessages(str, Enum):

        if not self.__doc__:
            self._set_standard_field_description()

    # TODO: Add test for get_model_class

    def get_model_class(self) -> ModelT:
        return self.__fields__.get(DATA_KEY).type_

    # TODO: Update _raise_no_model_exception() text. Model is now a requirement

    @staticmethod
    def _raise_no_model_exception() -> None:

    @classmethod
    def _create_subcls_inheriting_from_mixins_and_orig_cls(cls):

        # TODO: Refactor this, and possibly elsewhere in class

        def __init__(self, *args, **kwargs):
            # print(f'__init__ for obj of class: {self.__class__.__name__}')

        self._serializer_registry = self._create_serializer_registry()

    def _create_serializer_registry(self):
        # TODO: store in runtime, to remove dependencies
        registry = SerializerRegistry()

        registry.register(PandasDatasetToTarFileSerializer)

    FINISHED = 3


# TODO: Add 'apply' state
# TODO: Add 'failed' state and error management
# TODO: Consider the need for a 'waiting' state


class RunStateLogMessages(str, Enum):

            raise


# TODO: Change JobBase and friends into Generics such as one can annotated with
#       e.g. 'TaskTemplate[[int], int]' instead of just 'TaskTemplate'

JobBase.accept_mixin(NameJobBaseMixin)

    def __init__(self, *args: object, name: Optional[str] = None, **kwargs: object):
        # super().__init__()

        # TODO: refactor using state machine

        if not isinstance(self, JobTemplate) and not isinstance(self, Job):
            raise JobStateException('JobBase and subclasses not inheriting from JobTemplate '

        return getattr(self.objects, engine_choice)

    def _new_engine_config_if_new_cls(self, engine: IsEngine, engine_choice: EngineChoice) -> None:
        # TODO: when parsing config from file is implemented, make sure that the new engine
        #       config classes here reparse the config files
        engine_config_cls = engine.get_config_cls()
        if self._get_engine_config(engine_choice).__class__ is not engine_config_cls:

    ...


# TODO: Reimplement typing of flows and task to harmonize with new structure
def linear_flow_template_callable_decorator_cls(
        cls: Type['LinearFlowTemplate']
) -> IsTaskTemplatesFlowTemplateCallable['LinearFlowTemplate']:

        return TaskTemplate


# TODO: Would we need the possibility to refine task templates by adding new task parameters?

                                        JsonModel,
                                        JsonNestedDictsModel)

# TODO: switch from plural to singular for names of modules in omnipy modules
# TODO: call omnipy modules something else than modules, to distinguish from Python modules.
#       Perhaps plugins?

JsonModelT = TypeVar('JsonModelT', bound=Union[JsonModel, JsonListModel, JsonDictModel])

        return bool(other_job_same_unique_name) and id(other_job_same_unique_name) != id(job)

    def _update_job_registration(self, job: IsJob, state: RunState) -> None:
        # TODO: Reimplement logic using a state machine, e.g. "transitions" package
        if self._other_job_registered_with_same_unique_name(job):
            while self._other_job_registered_with_same_unique_name(job):
                job.regenerate_unique_name()

        if not self.__doc__:
            self._set_standard_field_description()

    # TODO: Add test for get_model_class

    def get_model_class(self) -> ModelT:
        return self.__fields__.get(DATA_KEY).type_

    # TODO: Update _raise_no_model_exception() text. Model is now a requirement

    @staticmethod
    def _raise_no_model_exception() -> None: