cursus.api.dag

Pipeline DAG API module.

This module provides the core DAG classes for building and managing pipeline topologies with intelligent dependency resolution.

class PipelineDAG(nodes=None, edges=None, strict=False)

Bases: object

Represents a pipeline topology as a directed acyclic graph (DAG). Each node is a step name; edges define dependencies.

Node declaration vs. auto-creation: a node is “declared” when it is passed in the nodes= constructor arg or added via add_node(). By default add_edge() auto-creates any endpoint that was not declared (lenient mode) — convenient, but it means a single typo in an edge name (add_edge("A", "TabularPreprocessing_traning")) silently spawns a phantom, unconfigured node and orphans the real one, and construction never raises. Two guards exist:

• validate_node_declarations() reports every edge endpoint that was never declared (always available, non-fatal) — call it to surface likely typos / forgotten add_node.

• strict=True turns the same condition into an immediate ValueError at add_edge (and constructor) time, so undeclared endpoints can never enter the graph.

add_edge(src, dst)

Add a directed edge from src to dst.

Lenient (default): auto-creates either endpoint if it is not yet a node — but does NOT mark it declared, so validate_node_declarations will still surface it as a likely typo. strict=True: raises ValueError if either endpoint was never declared via add_node.

add_node(node)

Add a single node to the DAG. This DECLARES the node (see class docstring).

get_dependencies(node)

Return immediate dependencies (parents) of a node.

topological_sort()

Return nodes in topological order.

Tolerates edges whose endpoints are not in nodes (dangling edges) by ignoring the unknown endpoints here rather than raising an opaque KeyError — structural problems like dangling edges are surfaced with clear messages by the serializer’s validation layer.

validate_node_declarations()

Return edge endpoints that were never explicitly declared via add_node / the nodes= arg.

Because add_edge auto-creates missing endpoints (lenient mode), a typo in an edge name silently produces a phantom, unconfigured node — and the serializer’s dangling-edge check cannot catch it, since add_edge has already promoted the typo into nodes. This is the only reliable detector: it compares every edge endpoint against the DECLARED set. An empty list means every edge endpoint was declared; any member is a likely typo or a forgotten add_node. Non-fatal — strict=True turns the same condition into a raise at add_edge time.

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_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

get_dependent_steps(step_name)

Get steps that depend on the given step.

get_step_dependencies(step_name)

Get immediate dependencies for a 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

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

Bases: BaseModel

Execution plan for pipeline with topological ordering.

model_config: ClassVar[ConfigDict] = {}

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

execution_order: List[str]

step_configs: Dict[str, dict]

dependencies: Dict[str, List[str]]

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

class PipelineDAGWriter(dag, metadata=None)

Bases: object

Writer for serializing PipelineDAG to various formats.

Supports: - JSON format with metadata - Pretty printing for readability - Validation before writing - Metadata tracking (creation time, etc.)

to_dict()

Convert PipelineDAG to dictionary representation.

Returns:

Dictionary with complete DAG representation including metadata

Return type:

Dict[str, Any]

to_json(pretty=True, indent=2)

Convert PipelineDAG to JSON string.

Parameters:

• pretty (bool) – If True, format JSON with indentation

• indent (int) – Number of spaces for indentation (if pretty=True)

Returns:

JSON string representation of the DAG

Return type:

str

write_to_file(filepath, pretty=True, indent=2, validate=True)

Write PipelineDAG to a JSON file.

Parameters:

• filepath (str | Path) – Path to output file

• pretty (bool) – If True, format JSON with indentation

• indent (int) – Number of spaces for indentation

• validate (bool) – If True, validate DAG before writing

Raises:

• ValueError – If DAG validation fails

• IOError – If file cannot be written

class PipelineDAGReader

Bases: object

Reader for deserializing PipelineDAG from various formats.

Supports: - JSON format - Validation during reading - Metadata extraction

classmethod extract_metadata(filepath)

Extract metadata from a DAG file without loading the full DAG.

Parameters:

filepath (str | Path) – Path to input file

Returns:

Dictionary containing metadata

Return type:

Dict[str, Any]

classmethod from_dict(data, validate=True)

Create PipelineDAG from dictionary representation.

Parameters:

• data (Dict[str, Any]) – Dictionary containing DAG data

• validate (bool) – If True, validate data before creating DAG

Returns:

PipelineDAG instance

Raises:

ValueError – If data is invalid or version is unsupported

Return type:

PipelineDAG

classmethod from_json(json_str, validate=True)

Create PipelineDAG from JSON string.

Parameters:

• json_str (str) – JSON string representation of DAG

• validate (bool) – If True, validate data before creating DAG

Returns:

PipelineDAG instance

Raises:

ValueError – If JSON is invalid or data is malformed

Return type:

PipelineDAG

classmethod read_from_file(filepath, validate=True)

Read PipelineDAG from a JSON file.

Parameters:

• filepath (str | Path) – Path to input file

• validate (bool) – If True, validate data before creating DAG

Returns:

PipelineDAG instance

Raises:

• FileNotFoundError – If file doesn’t exist

• ValueError – If file content is invalid

Return type:

PipelineDAG

export_dag_to_json(dag, filepath, metadata=None, pretty=True)

Convenience function to export a PipelineDAG to JSON file.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance to export

• filepath (str | Path) – Output file path

• metadata (Dict[str, Any] | None) – Optional metadata to include

• pretty (bool) – If True, format JSON with indentation

import_dag_from_json(filepath)

Convenience function to import a PipelineDAG from JSON file.

Parameters:

filepath (str | Path) – Input file path

Returns:

PipelineDAG instance

Return type:

PipelineDAG

Modules

base_dag
pipeline_dag_resolver	Pipeline DAG resolver for execution planning.
pipeline_dag_serializer	Pipeline DAG Serialization Module.