cursus.api

Cursus API module.

This module provides the main API interfaces for Cursus functionality, including DAG management and pipeline compilation.

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

Modules

dag	Pipeline DAG API module.
factory	DAG Configuration Factory API