import importlib
from abc import abstractmethod
from dataclasses import dataclass, field
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from typing import (
Any,
Dict,
Generic,
Iterator,
List,
Optional,
OrderedDict,
Protocol,
Tuple,
Type,
TypeAlias,
TypeVar,
Union,
)
from py_app_dev.core.config import ConfigElement
from py_app_dev.core.exceptions import UserNotificationException
from py_app_dev.core.runnable import Runnable
from .execution_context import ExecutionContext
[docs]
@dataclass
class PipelineStepConfig(ConfigElement):
#: Step name or class name if file is not specified
step: Optional[str] = None
#: Path to another pypeline file whose steps are spliced in at this position (instead of a step)
include: Optional[str] = None
#: Path to file with step class
file: Optional[str] = None
#: Python module with step class
module: Optional[str] = None
#: Step class name
class_name: Optional[str] = None
#: Command to run. For simple steps that don't need a class.
#:
#: Single command::
#:
#: run: ruff check .
#:
#: Or as list::
#:
#: run: [pytest, -v, --cov]
#:
#: Multiple commands (GitHub Actions style)::
#:
#: run: |
#: ruff check .
#: pytest -v --cov
run: Optional[Union[str, List[str]]] = None
#: Step description
description: Optional[str] = None
#: Step timeout in seconds
timeout_sec: Optional[int] = None
#: Custom step configuration
config: Optional[Dict[str, Any]] = None
#: Output group taken from the file where the step is *defined*, not the file that includes
#: it, so the step's output directory is identical whether its file is run standalone or
#: spliced into a larger pipeline. Assembly metadata: set during loading, never serialized.
home_group: Optional[str] = field(default=None, compare=False, repr=False)
_home_group_set: bool = field(default=False, compare=False, repr=False)
def __post_serialize__(self, d: Dict[str, Any]) -> Dict[str, Any]:
d = super().__post_serialize__(d)
d.pop("home_group", None)
d.pop("_home_group_set", None)
return d
[docs]
def set_home_group(self, group_name: Optional[str]) -> None:
"""Pin the output group to the file where this step is defined (idempotent per step)."""
self.home_group = group_name
self._home_group_set = True
[docs]
def resolve_output_group(self, fallback: Optional[str]) -> Optional[str]:
"""Group driving the step's output dir: its defining-file group, or `fallback` if not loaded from a file."""
return self.home_group if self._home_group_set else fallback
PipelineConfig: TypeAlias = Union[List[PipelineStepConfig], OrderedDict[str, List[PipelineStepConfig]]]
TPipelineStep = TypeVar("TPipelineStep", covariant=True)
@dataclass
class PipelineStepReference(Generic[TPipelineStep]):
"""Once a Step is found, keep the Step class reference to be able to instantiate it later."""
group_name: Optional[str]
_class: Type[TPipelineStep]
config: Optional[Dict[str, Any]] = None
@property
def name(self) -> str:
return self._class.__name__
class PipelineConfigIterator:
"""
Iterates over the pipeline configuration, yielding group name and steps configuration.
This class abstracts the iteration logic for PipelineConfig, which can be:
- A list of steps (group name is None)
- An OrderedDict with group names as keys and lists of steps as values.
The iterator yields tuples of (group_name, steps).
"""
def __init__(self, pipeline_config: PipelineConfig) -> None:
self._items = pipeline_config.items() if isinstance(pipeline_config, OrderedDict) else [(None, pipeline_config)]
def __iter__(self) -> Iterator[Tuple[Optional[str], List[PipelineStepConfig]]]:
"""Return an iterator."""
yield from self._items
class StepClassFactory(Generic[TPipelineStep], Protocol):
def create_step_class(self, step_config: PipelineStepConfig, project_root_dir: Path) -> Type[TPipelineStep]: ...
[docs]
class PipelineLoader(Generic[TPipelineStep]):
def __init__(self, pipeline_config: PipelineConfig, project_root_dir: Path, step_class_factory: Optional[StepClassFactory[TPipelineStep]] = None) -> None:
self.pipeline_config = pipeline_config
self.project_root_dir = project_root_dir
self.step_class_factory = step_class_factory
[docs]
def load_steps_references(self) -> List[PipelineStepReference[TPipelineStep]]:
result = []
for group_name, steps_config in PipelineConfigIterator(self.pipeline_config):
result.extend(self._load_steps(group_name, steps_config, self.project_root_dir, self.step_class_factory))
return result
@staticmethod
def _load_steps(
group_name: Optional[str], steps_config: List[PipelineStepConfig], project_root_dir: Path, step_class_factory: Optional[StepClassFactory[TPipelineStep]] = None
) -> List[PipelineStepReference[TPipelineStep]]:
result = []
for step_config in steps_config:
step_class_name = step_config.class_name or step_config.step
if not step_class_name:
raise UserNotificationException("A pipeline step must define a 'step' name. Please check your pipeline configuration.")
if step_config.module:
step_class = PipelineLoader[TPipelineStep]._load_module_step(step_config.module, step_class_name)
elif step_config.file:
step_class = PipelineLoader[TPipelineStep]._load_user_step(project_root_dir.joinpath(step_config.file), step_class_name)
else:
if step_class_factory:
step_class = step_class_factory.create_step_class(step_config, project_root_dir)
else:
raise UserNotificationException(
f"Step '{step_class_name}' has no 'module' nor 'file' defined nor a custom step class factory was provided. Please check your pipeline configuration."
)
# The output group comes from where the step is DEFINED, not the (possibly including) file
# being assembled here; this keeps a step's output dir stable across standalone vs included runs.
result.append(PipelineStepReference(step_config.resolve_output_group(group_name), step_class, step_config.config))
return result
@staticmethod
def _load_user_step(python_file: Path, step_class_name: str) -> Type[TPipelineStep]:
# Create a module specification from the file path
spec = spec_from_file_location(f"user__{python_file.stem}", python_file)
if spec and spec.loader:
step_module = module_from_spec(spec)
# Import the module
spec.loader.exec_module(step_module)
try:
step_class = getattr(step_module, step_class_name)
except AttributeError:
raise UserNotificationException(f"Could not load class '{step_class_name}' from file '{python_file}'. Please check your pipeline configuration.") from None
return step_class
raise UserNotificationException(f"Could not load file '{python_file}'. Please check the file for any errors.")
@staticmethod
def _load_module_step(module_name: str, step_class_name: str) -> Type[TPipelineStep]:
try:
module = importlib.import_module(module_name)
step_class = getattr(module, step_class_name)
except ImportError:
raise UserNotificationException(f"Could not load module '{module_name}'. Please check your pipeline configuration.") from None
except AttributeError:
raise UserNotificationException(f"Could not load class '{step_class_name}' from module '{module_name}'. Please check your pipeline configuration.") from None
return step_class
TExecutionContext = TypeVar("TExecutionContext", bound=ExecutionContext)
[docs]
class PipelineStep(Generic[TExecutionContext], Runnable):
"""One can create subclasses of PipelineStep that specify the type of ExecutionContext they require."""
def __init__(self, execution_context: TExecutionContext, group_name: Optional[str], config: Optional[Dict[str, Any]] = None) -> None:
super().__init__(self.get_needs_dependency_management())
self.execution_context = execution_context
self.group_name = group_name
self.config = config
self.project_root_dir = self.execution_context.project_root_dir
@property
def output_dir(self) -> Path:
output_dir = self.execution_context.create_artifacts_locator().build_dir
if self.group_name:
output_dir = output_dir / self.group_name
return output_dir
[docs]
@abstractmethod
def update_execution_context(self) -> None:
"""
Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.
A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.
"""
pass
[docs]
def get_needs_dependency_management(self) -> bool:
"""If false, the step executor will not check for outdated dependencies. This is useful for steps consisting of command lines which shall always run."""
return True
