Execution

DAG resolution, shell backends, caching, and template rendering for tasks.

DAG resolution and topological sort for task dependencies.

conda_workspaces.graph.resolve_execution_order(target: str, tasks: dict[str, Task], *, skip_deps: bool = False) → list[str]

Return task names in the order they should execute to run target.

Uses graphlib.TopologicalSorter. Only the transitive closure of target’s dependencies is included – unrelated tasks are omitted.

Raises TaskNotFoundError if target or any dependency is missing. Raises CyclicDependencyError if the dependency graph has a cycle.

Shell execution backend for running task commands.

class conda_workspaces.runner.SubprocessShell

Execute shell commands, optionally inside an activated conda env.

When conda_prefix is given the command is executed inside an activated conda environment (mirroring conda run). Otherwise the command runs directly in the current shell.

run(cmd: str | list[str], env: dict[str, str], cwd: Path, conda_prefix: Path | None = None, clean_env: bool = False) → int

Execute cmd and return the process exit code.

Task output caching using file fingerprints.

Cache entries are stored in a platform-appropriate directory via platformdirs. Each task base directory gets a subdirectory keyed by a hash of its path. This is the manifest directory, or the invocation directory when only user tasks are available. Within that, each task has a JSON file containing fingerprints of its inputs and outputs.

SHA-256 digests are the file identity.

conda_workspaces.cache.is_cached(project_root: Path, task_name: str, cmd: str | list[str], env: dict[str, str], input_patterns: list[str], output_patterns: list[str], cwd: Path, *, conda_prefix: Path | None = None) → bool

Check whether the task can be skipped (cache hit).

project_root is the manifest directory, or the invocation directory when only user tasks are available. The keyword name is retained for compatibility.

Returns True only when all of the following hold:

1. A cache entry exists for the task.

2. The command and env hashes match.

3. All input files match by SHA-256 digest.

4. All output files still exist and match.

conda_workspaces.cache.save_cache(project_root: Path, task_name: str, cmd: str | list[str], env: dict[str, str], input_patterns: list[str], output_patterns: list[str], cwd: Path, *, conda_prefix: Path | None = None) → None

Write or update the cache entry for a task.

project_root is the manifest directory, or the invocation directory when only user tasks are available. The keyword name is retained for compatibility.

Jinja2 template rendering for task commands and paths.

conda_workspaces.template.render(template_str: str, manifest_path: Path | None = None, task_args: dict[str, str] | None = None, extra_context: dict[str, object] | None = None, target_prefix: Path | None = None) → str

Render a Jinja2 template string with the conda-workspaces template context.

If template_str contains no template markers it is returned as-is (fast path that avoids Jinja2 import entirely).

conda_workspaces.template.render_command(template_str: str, manifest_path: Path | None = None, task_args: dict[str, str] | None = None, target_prefix: Path | None = None) → str

Render a shell command template with task arguments shell-quoted.

Command strings are executed through the native shell, so named task arguments are data values that must occupy one shell word when they are interpolated.

conda_workspaces.template.render_list(items: list[str], manifest_path: Path | None = None, task_args: dict[str, str] | None = None, target_prefix: Path | None = None) → list[str]

Render each string in items through the template engine.