import inspect from typing import Dict, Iterable, Set, Type from .aggregates import * from .base import BaseStep, StepError from .preparation import * from .search import * from .upload_cms_data import * from .utils import * from .validation import * __all__ = ["steps", "get_step_depends_on"] def __get_nested_steps(cls) -> Iterable[Type[BaseStep]]: for subclass in cls.__subclasses__(): if not inspect.isabstract(subclass): yield subclass yield from __get_nested_steps(subclass) __steps_map: Dict[str, Type[BaseStep]] = {c.step_name: c for c in __get_nested_steps(BaseStep)} steps: Iterable[Type[BaseStep]] = __steps_map.values() def get_step_depends_on(step_cls: Type[BaseStep]) -> Set[Type[BaseStep]]: depends_on = set() for parent in step_cls.depends_on: if isinstance(parent, str): if parent in __steps_map: parent = __steps_map[parent] else: raise StepError(f"Can't find step '{parent}'") depends_on.add(parent) return depends_on