cursus.core.utils

Core utilities for cursus.

This module provides utility functions and classes that support the core functionality of cursus, including path resolution, configuration management, and other common operations.

class HybridPathResolver

Bases: object

Hybrid path resolver that works across all deployment scenarios.

This class implements the core hybrid resolution algorithm that uses Package Location Discovery first, then Working Directory Discovery as fallback.

resolve_path(project_root_folder, relative_path)

Hybrid path resolution: Package location first, then working directory discovery.

This method implements the core hybrid resolution algorithm that works across all deployment scenarios: - Lambda/MODS bundled: Package location discovery - Development monorepo: Monorepo structure detection - Pip-installed separated: Working directory discovery fallback

Parameters:

• project_root_folder (str) – Root folder name for the user’s project

• relative_path (str) – Relative path from project root to target directory/file

Returns:

Resolved absolute path if found, None otherwise

Return type:

str | None

resolve_hybrid_path(project_root_folder, relative_path)

Convenience function for hybrid path resolution.

Parameters:

• project_root_folder (str) – Root folder name for the user’s project

• relative_path (str) – Relative path from project root to target

Returns:

Resolved absolute path if found, None otherwise

Return type:

str | None

get_hybrid_resolution_metrics()

Get current hybrid resolution performance metrics.

class HybridResolutionConfig

Bases: object

Configuration for hybrid resolution rollout.

static get_hybrid_resolution_mode()

Get hybrid resolution mode: ‘full’, ‘fallback_only’, ‘disabled’.

static is_hybrid_resolution_enabled()

Check if hybrid resolution is enabled via environment variable.

set_project_root(project_root)

Push the project folder for caller-hook (Strategy 0) path resolution.

Accepts either a project directory (Path(__file__).parent) or a file (__file__) — both are normalized to the project root via resolve_anchor(), so callers may push whichever they have on hand.

Parameters:

project_root (str | Path | None) – Path to the user’s project folder or a file inside it (typically Path(__file__).parent or __file__ of the module defining generate_pipeline()), or None to clear it.

get_project_root()

Return the currently pushed project root (caller-hook anchor), or None.

resolve_anchor(anchor)

Normalize a file-or-directory anchor into a project-root path string.

This is the single shared normalizer used by every entry point (the DAG compiler, ExecutionDocumentGenerator, DAGConfigFactory, the scaffold, …) so anchor_file=__file__ and project_root=<dir> collapse to the same project root and resolve identically everywhere.

• A file (e.g. __file__ of the module defining generate_pipeline()) resolves to its parent directory — the project folder.

• A directory (e.g. Path(__file__).parent) resolves to itself.

• A path that does not exist yet but is “file-shaped” (has a suffix) is treated as a file and reduced to its parent, so anchor_file=__file__ works even before the module is materialized on disk.

Parameters:

anchor (str | Path | None) – A file path, a directory path, or None.

Returns:

Absolute path string of the project root, or None if anchor is falsy.

Return type:

str | None

find_project_folder_generic(project_root_folder, max_depth_up=5, max_depth_down=3, reference_points=None)

Generic algorithm to find a uniquely named project folder.

Searches from multiple reference points in both upward and downward directions, using recursive traversal to handle nested directory structures.

Parameters:

• project_root_folder (str) – Name or path of project folder (e.g., “atoz_xgboost” or “projects/atoz_xgboost”)

• max_depth_up (int) – Maximum levels to search upward (default: 5)

• max_depth_down (int) – Maximum levels to search downward (default: 3)

• reference_points (List[Path] | None) – Optional list of starting points for search. If None, uses [Path.cwd(), cursus_package_root]

Returns:

Absolute path to project folder if found, None otherwise

Return type:

Path | None

Example

>>> find_project_folder_generic("atoz_xgboost")
PosixPath('/Users/user/workspace/projects/atoz_xgboost')

>>> find_project_folder_generic("projects/atoz_xgboost")
PosixPath('/Users/user/workspace/projects/atoz_xgboost')

get_generic_discovery_metrics()

Get current generic path discovery performance metrics.

discover_pipeline_projects(root=None, names=None)

Discover Cursus pipeline projects under root (or locate specific ones by name).

Parameters:

• root (str | Path | None) – Directory to scan for project subdirectories. When None and names is given, each name is located via cursus.core.utils.find_project_folder_generic() (cross-deployment search). When both are None, returns an empty list (nothing to scan).

• names (List[str] | None) – Optional explicit list of project folder names. If given, only these are returned (located under root if provided, else via generic discovery).

Returns:

A list of ProjectInfo, one per recognized pipeline project, sorted by name. Directories that are not pipeline projects (no config dir) are skipped.

Return type:

List[ProjectInfo]

summarize_project(project_dir)

Summarize a single pipeline-project directory.

Returns a ProjectInfo, or None if project_dir is not a recognizable Cursus pipeline project (no pipeline_config/pipeline_configs directory).

class ProjectInfo(name, path, config_dir=None, config_files=<factory>, has_dockers=False, has_scripts=False, pipeline_modules=<factory>)

Bases: object

Summary of one discovered pipeline project.

config_dir: str | None = None

property config_file_count: int

property distinct_config_types: List[str]

Union of config class names referenced across all of the project’s configs.

has_dockers: bool = False

has_scripts: bool = False

to_dict()

name: str

path: str

config_files: List[ConfigSummary]

pipeline_modules: List[str]

class ConfigSummary(file, node_count=0, config_types=<factory>, nodes=<factory>, created_at=None, error=None)

Bases: object

Summary of one config JSON file inside a project.

created_at: str | None = None

error: str | None = None

node_count: int = 0

to_dict()

file: str

config_types: Dict[str, str]

nodes: List[str]

install_nvme_aware_security_patch()

Patch MODSWorkflowHelper to be NVMe-aware for both training and processing steps.

For NVMe instances, clears volume_kms_key so it’s excluded from the API request. Uses hybrid logic: config override (processor._skip_volume_kms) > auto-detect from instance_supports_kms(processor.instance_type).

Idempotent — safe to call multiple times. Returns True if the patch was installed (or was already installed), False if dependencies are unavailable.

Modules

generic_path_discovery	Generic Path Discovery Algorithm for Project Folders.
hybrid_path_resolution	Hybrid Path Resolution System for Cursus.
nvme_security	NVMe-aware security patching for SageMaker training and processing steps.
project_discovery	Discover and summarize Cursus pipeline projects under a root directory.