"""Workspace context — lazy properties for conda & workspace state.
Provides a namespace of lazily-evaluated properties that downstream
code can use without importing conda at module level. This keeps
import-time overhead negligible.
"""
from __future__ import annotations
import os
import stat
from contextlib import contextmanager
from pathlib import Path
from threading import RLock
from typing import TYPE_CHECKING, cast
from .exceptions import CondaWorkspacesError, EnvironmentNameInvalidError
from .paths import is_path_segment
if TYPE_CHECKING:
from collections.abc import Iterator
from conda.models.environment import Environment
from .models import WorkspaceConfig
_package_cache_isolated = False
_package_cache_lock = RLock()
[docs]
@contextmanager
def isolated_package_cache(enabled: bool) -> Iterator[None]:
"""Route conda package-cache writes to disposable storage when enabled.
Conda solvers may populate ``context.pkgs_dirs`` while resolving, before
any install transaction runs. Dry-run solver paths copy cached repodata
into disposable storage while retaining configured package caches as
read sources.
"""
if not enabled:
yield
return
global _package_cache_isolated
with _package_cache_lock:
if _package_cache_isolated:
yield
return
import shutil
import tempfile
from conda.base.context import context
configured_caches = context.pkgs_dirs
with tempfile.TemporaryDirectory(prefix="conda-workspaces-pkgs-") as cache_dir:
scratch_cache = Path(cache_dir)
for configured_cache in configured_caches:
repodata_cache = Path(configured_cache) / "cache"
if repodata_cache.is_dir():
shutil.copytree(repodata_cache, scratch_cache / "cache")
break
_package_cache_isolated = True
try:
with context._override(
"_pkgs_dirs",
(scratch_cache, *configured_caches),
):
yield
finally:
_package_cache_isolated = False
[docs]
class WorkspaceContext:
"""Lazy-evaluated context for the current workspace.
Properties are resolved on first access and cached. Conda imports
are deferred to keep plugin load time under 1 ms.
"""
def __init__(self, config: WorkspaceConfig | None = None) -> None:
self._config = config
self._cache: dict[str, object] = {}
@property
def config(self) -> WorkspaceConfig:
"""The parsed workspace configuration."""
if self._config is None:
from .manifests import detect_and_parse
_, self._config = detect_and_parse()
return self._config
@property
def root(self) -> Path:
"""Workspace root directory."""
return Path(self.config.root)
@property
def envs_dir(self) -> Path:
"""Directory where project-local environments are stored."""
if "envs_dir" not in self._cache:
root = self.root
envs_dir = root / self.config.envs_dir
candidates: list[Path] = []
candidate = envs_dir
while candidate != root and candidate != candidate.parent:
candidates.append(candidate)
candidate = candidate.parent
for candidate in candidates:
if candidate.is_symlink():
raise CondaWorkspacesError(
f"Workspace environments path contains a symlink: {candidate}"
)
resolved_root = root.resolve(strict=False)
resolved = envs_dir.resolve(strict=False)
try:
relative = resolved.relative_to(resolved_root)
except ValueError as exc:
raise CondaWorkspacesError(
"Workspace environments directory escapes the workspace:"
f" {envs_dir}"
) from exc
if not relative.parts:
raise CondaWorkspacesError(
"Workspace environments directory cannot be the workspace root."
)
self._cache["envs_dir"] = envs_dir
self._cache["envs_dir_candidates"] = tuple(candidates)
envs_dir = cast("Path", self._cache["envs_dir"])
for candidate in cast(
"tuple[Path, ...]",
self._cache["envs_dir_candidates"],
):
if candidate.is_symlink():
raise CondaWorkspacesError(
f"Workspace environments path contains a symlink: {candidate}"
)
return envs_dir
[docs]
def envs_dir_identity(self) -> tuple[int, int] | None:
"""Return the current environments directory identity without following it."""
envs_dir = self.envs_dir
try:
current = envs_dir.lstat()
except FileNotFoundError:
return None
if not stat.S_ISDIR(current.st_mode):
raise CondaWorkspacesError(
f"Workspace environments path is not a directory: {envs_dir}"
)
return current.st_dev, current.st_ino
[docs]
def require_envs_dir_identity(self, expected: tuple[int, int]) -> None:
"""Reject replacement of the environments directory during an operation."""
current = self.envs_dir_identity()
if current != expected:
raise CondaWorkspacesError(
"Workspace environments directory changed while it was being used."
)
[docs]
def iter_installed_prefixes(self) -> Iterator[tuple[Path, tuple[int, int]]]:
"""Yield valid conda prefixes and the identities that were inspected."""
from conda.core.envs_manager import PrefixData
envs_dir = self.envs_dir
if not envs_dir.is_dir():
return
for prefix in envs_dir.iterdir():
try:
before = prefix.lstat()
except FileNotFoundError:
continue
if stat.S_ISLNK(before.st_mode):
raise CondaWorkspacesError(
f"Workspace environment prefix cannot be a symlink: {prefix}"
)
if (
not stat.S_ISDIR(before.st_mode)
or not PrefixData(str(prefix)).is_environment()
):
continue
try:
after = prefix.lstat()
except FileNotFoundError as exc:
raise CondaWorkspacesError(
"Workspace environment prefix changed while it was inspected:"
f" {prefix}"
) from exc
identity = before.st_dev, before.st_ino
if (
not stat.S_ISDIR(after.st_mode)
or (
after.st_dev,
after.st_ino,
)
!= identity
):
raise CondaWorkspacesError(
"Workspace environment prefix changed while it was inspected:"
f" {prefix}"
)
yield prefix, identity
@property
def platform(self) -> str:
"""Current conda subdir (e.g. ``osx-arm64``)."""
if "platform" not in self._cache:
from conda.base.context import context
self._cache["platform"] = context.subdir
return cast("str", self._cache["platform"])
@property
def root_prefix(self) -> Path:
"""Conda root prefix (base environment)."""
if "root_prefix" not in self._cache:
from conda.base.context import context
self._cache["root_prefix"] = Path(context.root_prefix)
return cast("Path", self._cache["root_prefix"])
@property
def is_ci(self) -> bool:
"""Whether the process is running in a CI environment."""
return os.environ.get("CI", "").lower() in ("true", "1", "yes")
[docs]
@staticmethod
def validate_environment_name(env_name: str) -> None:
"""Reject an environment name that cannot map to a local prefix."""
if not is_path_segment(env_name):
raise EnvironmentNameInvalidError(env_name)
[docs]
def env_prefix(self, env_name: str) -> Path:
"""Return the prefix path for a named environment."""
self.validate_environment_name(env_name)
prefix = self.envs_dir / env_name
if prefix.is_symlink():
raise CondaWorkspacesError(
f"Workspace environment prefix cannot be a symlink: {prefix}"
)
return prefix
[docs]
def env_exists(self, env_name: str) -> bool:
"""Check whether the prefix is a valid conda environment."""
from conda.core.envs_manager import PrefixData
prefix = self.env_prefix(env_name)
return PrefixData(str(prefix)).is_environment()
[docs]
def envs_from_manifest(
self,
env_name: str,
*,
requested_platforms: tuple[str, ...] = (),
) -> list[Environment]:
"""Build ``Environment`` objects from the workspace manifest.
Produces an :class:`~conda.models.environment.Environment` per
target platform with ``requested_packages`` populated from the
manifest's declared specs (no solver, no installed packages
required) — the novel capability of ``conda workspace export``
vs. ``conda export``, which always operates on an installed
prefix. When the manifest declares platforms conda doesn't
know, falls back to :attr:`platform` so the export still
produces something useful rather than crashing on validation.
This is the natural entry point for third-party exporter
plugins (or other tooling) that want to turn a ``conda.toml``
into a list of :class:`Environment` objects without going
through the CLI.
"""
from .export import envs_from_manifest
return envs_from_manifest(
self, env_name, requested_platforms=requested_platforms
)
[docs]
def envs_from_prefix(
self,
env_name: str,
*,
requested_platforms: tuple[str, ...] = (),
from_history: bool = False,
no_builds: bool = False,
ignore_channels: bool = False,
) -> list[Environment]:
"""Build ``Environment`` objects from an installed workspace prefix.
Thin wrapper around the same :meth:`Environment.from_prefix` +
:meth:`Environment.extrapolate` pair that
:func:`conda.cli.main_export.execute` uses; the only
workspace-specific pieces are the prefix lookup
(:meth:`env_prefix`) and the
:class:`EnvironmentNotInstalledError` guard.
When *requested_platforms* is empty or equals
``(self.platform,)``, a single :class:`Environment` for the
host platform is returned. Otherwise one
:class:`Environment` per requested platform is produced via
:meth:`Environment.extrapolate`.
"""
from .export import envs_from_prefix
return envs_from_prefix(
self,
env_name,
requested_platforms=requested_platforms,
from_history=from_history,
no_builds=no_builds,
ignore_channels=ignore_channels,
)
[docs]
def envs_from_lockfile(
self,
env_name: str,
*,
requested_platforms: tuple[str, ...] = (),
) -> list[Environment]:
"""Load ``Environment`` objects from the workspace ``conda.lock``.
Delegates to :class:`~conda_workspaces.lockfile.CondaLockLoader`,
the same entry point conda uses when it reads ``--file
conda.lock`` through
:meth:`Environment.from_cli_with_file_envs`.
When *requested_platforms* is empty, every platform present in
the lockfile is returned. Otherwise the list is filtered and
:class:`PlatformError` is raised for any requested platform the
lockfile does not contain.
"""
from .export import envs_from_lockfile
return envs_from_lockfile(
self, env_name, requested_platforms=requested_platforms
)
[docs]
class CondaContext:
"""Lazy-evaluated namespace exposed as ``conda.*`` in task templates.
Attribute access is deferred so conda internals load only when a
template references a variable.
"""
def __init__(
self,
manifest_path: Path | None = None,
target_prefix: Path | None = None,
) -> None:
self._manifest_path = manifest_path
self._target_prefix = target_prefix
@property
def platform(self) -> str:
"""The conda platform/subdir string, e.g. ``linux-64`` or ``osx-arm64``."""
from conda.base.context import context
return context.subdir
@property
def environment_name(self) -> str:
"""Name of the selected or currently active conda environment."""
if self._target_prefix is not None:
return self._target_prefix.name
from conda.base.context import context
if context.active_prefix:
return Path(context.active_prefix).name
return "base"
@property
def environment(self) -> _EnvironmentProxy:
"""Allows ``{{ conda.environment.name }}`` in templates."""
return _EnvironmentProxy(self.environment_name)
@property
def prefix(self) -> str:
"""Absolute path to the target conda environment prefix."""
if self._target_prefix is not None:
return str(self._target_prefix)
from conda.base.context import context
return str(context.target_prefix)
@property
def version(self) -> str:
"""The installed conda version string."""
from conda import __version__
return __version__
@property
def manifest_path(self) -> str:
"""Path to the task definition file, or empty string if unknown."""
return str(self._manifest_path) if self._manifest_path else ""
@property
def init_cwd(self) -> str:
"""The working directory at the time of context creation."""
return os.getcwd()
@property
def is_win(self) -> bool:
"""True when running on Windows."""
from conda.base.constants import on_win
return on_win
@property
def is_unix(self) -> bool:
"""True when running on a Unix-like system (Linux or macOS)."""
from conda.base.constants import on_win
return not on_win
@property
def is_osx(self) -> bool:
"""True when the host platform is macOS."""
from conda.base.context import context
return context.platform == "osx"
@property
def is_linux(self) -> bool:
"""True when the host platform is Linux."""
from conda.base.context import context
return context.platform == "linux"
class _EnvironmentProxy:
"""Allows ``{{ conda.environment.name }}`` in templates."""
def __init__(self, name: str) -> None:
self.name = name
[docs]
def build_template_context(
manifest_path: Path | None = None,
task_args: dict[str, str] | None = None,
target_prefix: Path | None = None,
) -> dict[str, object]:
"""Build the full Jinja2 template context dict.
The returned dict contains:
- ``conda``: a :class:`CondaContext` instance
- ``pixi``: alias to the same context (for pixi.toml compatibility)
- Any user-supplied task argument values
"""
ctx = CondaContext(manifest_path=manifest_path, target_prefix=target_prefix)
result: dict[str, object] = {"conda": ctx, "pixi": ctx}
if task_args:
reserved = {"conda", "pixi"}
for key, value in task_args.items():
if key not in reserved:
result[key] = value
return result
