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 Kahn’s algorithm for topological sort. 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 project gets a subdirectory keyed by a hash of the project root path. Within that, each task has a JSON file containing fingerprints of its inputs and outputs.

A fast pre-check using (mtime, size) tuples avoids SHA-256 hashing when files haven’t been touched since the last run.

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

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

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 (mtime, size) – falling back to SHA-256 if the fast check fails.

4. All output files still exist and match.

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

Write or update the cache entry for a task.

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) → 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_list(items: list[str], manifest_path: Path | None = None, task_args: dict[str, str] | None = None) → list[str]

Render each string in items through the template engine.