cursus.api.dag.pipeline_dag_resolver

Pipeline DAG resolver for execution planning.

class PipelineExecutionPlan(*, execution_order, step_configs, dependencies, data_flow_map)

Bases: BaseModel

Execution plan for pipeline with topological ordering.

execution_order: List[str]

step_configs: Dict[str, dict]

dependencies: Dict[str, List[str]]

data_flow_map: Dict[str, Dict[str, str]]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class PipelineDAGResolver(dag, workspace_dirs=None, config_path=None, available_configs=None, metadata=None, validate_on_init=True)

Bases: object

Enhanced resolver with StepCatalog integration for reliable, deployment-agnostic DAG resolution.

create_execution_plan()

Create topologically sorted execution plan with optional step config resolution.

get_step_dependencies(step_name)

Get immediate dependencies for a step.

get_dependent_steps(step_name)

Get steps that depend on the given step.

validate_dag_integrity()

REFACTORED: Comprehensive DAG validation using StepCatalog.

IMPROVEMENTS: - Step existence validation using catalog - Component availability checking (builders, contracts, specs, configs) - Workspace compatibility validation - Enhanced error messages with suggestions

get_config_resolution_preview()

Get a preview of how DAG nodes would be resolved to configurations.

Returns:

Preview information if config resolver is available, None otherwise

Return type:

Dict[str, Any] | None