"""Lockfile generation, consumption and env-spec plugin for ``conda.lock``.
Single source of truth for every ``conda.lock`` concern. The file owns
the read path, the write path, the ``CondaEnvironmentSpecifier`` plugin
class (:class:`CondaLockLoader`), and the plugin metadata (name,
aliases, default filename) consumed by ``plugin.py`` and
:mod:`.export`.
The ``conda.lock`` format is a *derivative* of rattler-lock v6
(``pixi.lock``): same schema machinery, same top-level keys
(``version``, ``environments``, ``packages``), but with an on-disk
``version: 1`` byte that identifies the file as conda-workspaces-owned.
:class:`CondaLockLoader` shares rattler-lock v6 conversion models with
:mod:`conda_lockfiles.rattler_lock.v6`. The read path performs an in-memory
``version: 6`` swap before delegating YAML -> ``Environment`` conversion; the
write path uses the same public package model while preserving conda-workspaces'
multi-environment and external-package layout.
The file layout is::
version: 1
environments:
<name>:
channels: [{url: ...}, ...]
packages:
<platform>: [{conda: <url>}, ...]
packages:
- conda: <url>
sha256: ...
md5: ...
depends: [...]
...
On the *write* side, :func:`generate_lockfile` solves each environment
and delegates YAML serialisation to the ``multiplatform_export`` hook
in :mod:`.export` (the same path ``conda export`` uses), so every
``conda.lock`` on disk comes out of a single formatter. On the *read*
side, :func:`install_from_lockfile` extracts the package list for one
environment + platform via :class:`CondaLockLoader` and installs the
exact URLs, bypassing the solver entirely.
"""
from __future__ import annotations
import io
import os
import stat
import sys
import tempfile
from collections.abc import Mapping
from contextlib import contextmanager, nullcontext
from copy import deepcopy
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import TYPE_CHECKING, cast
from urllib.parse import unquote, urlsplit
from conda.common.io import dashlist
from conda.core.prefix_data import delete_prefix_from_linked_data
from conda.models.dist import Dist
from conda.plugins.types import EnvironmentSpecBase
from .context import isolated_package_cache
from .exceptions import (
AllTargetsUnsolvableError,
CondaWorkspacesError,
EnvironmentNotFoundError,
LockfileIntegrityError,
LockfileMergeError,
LockfileNotFoundError,
LockfileStaleError,
PlatformError,
SolveError,
)
from .models import (
LockfileStatus,
has_url_credentials,
has_url_credentials_in_data,
redact_channel_name,
redact_channel_url,
redact_url,
)
from .parsing import (
BoundedYamlLoader,
DocumentExpansionBudget,
DocumentStringBudget,
decode_limited_text,
document_values_equal,
plain_document,
read_limited_text,
validate_document_limits,
)
from .paths import (
anchored_directory,
atomic_write_text,
canonicalize_system_path_alias,
output_paths_collide,
parse_relative_posix_path,
read_regular_file_bytes,
regular_file_generation,
validate_directory_output,
validate_file_output,
)
if TYPE_CHECKING:
from collections.abc import Callable, Iterable, Iterator, Sequence
from typing import Any, ClassVar, Final, Literal, TypeAlias
from conda.common.path import PathType
from conda.core.link import PrefixSetup
from conda.models.environment import Environment, EnvironmentConfig
from conda.models.match_spec import MatchSpec
from conda.models.records import PackageRecord
from .context import WorkspaceContext
from .models import Environment as WorkspaceEnvironment
from .models import WorkspaceConfig
from .paths import FileGeneration
from .resolver import ResolvedEnvironment
#: On-disk lockfile format version. Distinct from the rattler-lock v6
#: schema version so that tools can tell a conda-workspaces-owned lock
#: from a pixi-owned one at a glance.
LOCKFILE_VERSION: Final = 1
UrlOrigin: TypeAlias = tuple[str, str, int | None, str | None, str | None]
UrlLocation: TypeAlias = tuple[UrlOrigin, tuple[str, ...]]
ChannelPathTree: TypeAlias = dict[str, "ChannelPathTree"]
ChannelLocations: TypeAlias = dict[UrlOrigin, ChannelPathTree]
#: The canonical lockfile filename.
LOCKFILE_NAME: Final = "conda.lock"
#: Canonical, versioned plugin name. Stable across schema bumps;
#: follows the ``conda-lockfiles`` alias policy (see
#: ``docs/reference/format-aliases.md``).
FORMAT: Final = "conda-workspaces-lock-v1"
#: User-friendly aliases. Unversioned names are convenience handles
#: that may migrate to a newer ``FORMAT`` in the future.
ALIASES: Final = ("conda-workspaces-lock", "workspace-lock")
#: Default filenames this plugin handles.
DEFAULT_FILENAMES: Final = (LOCKFILE_NAME,)
MAX_LOCKFILE_BYTES: Final = 128 * 1024**2
MAX_LOCKFILE_DEPTH: Final = 128
MAX_LOCKFILE_COLLECTION_ITEMS: Final = 100_000
MAX_LOCKFILE_ITEMS: Final = 1_000_000
MAX_LOCKFILE_MATCH_ATTEMPTS: Final = 100_000
MAX_LOCKFILE_BATCH_MATCH_ATTEMPTS: Final = MAX_LOCKFILE_ITEMS
MAX_LOCKFILE_BATCH_TARGETS: Final = 10_000
MAX_LOCKFILE_BATCH_PACKAGE_REFS: Final = MAX_LOCKFILE_COLLECTION_ITEMS
MAX_LOCKFILE_BATCH_EXPANDED_ITEMS: Final = MAX_LOCKFILE_ITEMS
MAX_LOCKFILE_DECLARATION_ITEMS: Final = MAX_LOCKFILE_COLLECTION_ITEMS
_CURRENT_LOCKFILE_OUTPUT_GENERATION = object()
@dataclass
class _SolvedEnvironment:
"""Solved lockfile row that can carry a Pixi rich-platform name."""
name: str
platform: str
package_platform: str
config: EnvironmentConfig
explicit_packages: Sequence[PackageRecord]
external_packages: dict[str, list[str]] = field(default_factory=dict)
[docs]
def load_lockfile_data(
content: str | bytes,
*,
string_budget: DocumentStringBudget | None = None,
) -> dict[str, Any]:
"""Parse in-memory lockfile YAML with the same safe loader as disk reads."""
text = decode_limited_text(
content,
maximum_bytes=MAX_LOCKFILE_BYTES,
label="Lockfile YAML",
)
data = BoundedYamlLoader(
label="Lockfile YAML",
maximum_bytes=MAX_LOCKFILE_BYTES,
maximum_depth=MAX_LOCKFILE_DEPTH,
maximum_collection_items=MAX_LOCKFILE_COLLECTION_ITEMS,
maximum_items=MAX_LOCKFILE_ITEMS,
).load(text, string_budget=string_budget)
if not isinstance(data, Mapping):
raise ValueError("Lockfile YAML must contain a mapping")
return dict(data)
def load_lockfile_path(path: Path) -> dict[str, Any]:
"""Read and parse one lockfile under explicit resource limits."""
content = read_limited_text(
path,
maximum_bytes=MAX_LOCKFILE_BYTES,
label="Lockfile YAML",
)
return load_lockfile_data(content)
def lockfile_path(ctx: WorkspaceContext) -> Path:
"""Return the path to the workspace lockfile (``<root>/conda.lock``)."""
return ctx.root / LOCKFILE_NAME
def _normalize_channel_urls(urls: Iterable[str]) -> list[str]:
"""Strip trailing slashes from channel URLs for comparison."""
return [url.rstrip("/") for url in urls]
def _manifest_channel_entries(
config: WorkspaceConfig,
environment: WorkspaceEnvironment,
) -> list[dict[str, str]]:
"""Return manifest channel entries without conda canonical-name deduplication."""
seen: set[str] = set()
entries: list[dict[str, str]] = []
for ch in config.channels:
url = redact_channel_url(ch)
normalized = url.rstrip("/")
if normalized not in seen:
seen.add(normalized)
entries.append({"url": url})
for feature in config.resolve_features(environment):
for ch in feature.channels:
url = redact_channel_url(ch)
normalized = url.rstrip("/")
if normalized not in seen:
seen.add(normalized)
entries.append({"url": url})
return entries
def lockfile_status(
ctx: WorkspaceContext,
config: WorkspaceConfig,
*,
platform: str | None = None,
) -> LockfileStatus:
"""Determine the lockfile status relative to the workspace manifest."""
lock = lockfile_path(ctx)
if not lock.is_file():
return LockfileStatus(status=LockfileStatus.MISSING)
data = load_lockfile_path(lock)
return check_lockfile_satisfiability(config, data, platform or ctx.platform)
[docs]
def check_lockfile_satisfiability(
config: WorkspaceConfig,
lockfile_data: dict[str, Any] | CondaLockLoader,
current_platform: str,
*,
environment: str | None = None,
_validated_envs: Mapping[str, ResolvedEnvironment] | None = None,
_match_budget: list[int] | None = None,
_total_match_budget: list[int] | None = None,
_batch_expansion_budget: DocumentExpansionBudget | None = None,
_channel_locations_cache: dict[tuple[int, str], ChannelLocations] | None = None,
_resolution_budget: list[int] | None = None,
_exact_target: bool = False,
) -> LockfileStatus:
"""Check whether *lockfile_data* satisfies the manifest's requirements.
Returns a :class:`LockfileStatus` with ``status=UP_TO_DATE`` when
the lockfile covers every environment, platform, channel, and
dependency declared in *config*. Returns ``status=OUT_OF_DATE``
with a human-readable *reason* otherwise.
Set *environment* to check package requirements only for that environment
on *current_platform*. Environment names, channels and declared platforms
are still checked across the complete workspace. Use
:meth:`CondaLockLoader.check_targets` to check several exact targets after
validating those declarations once.
"""
from .resolver import MAX_RESOLUTION_WORK_ITEMS
_stale = LockfileStatus.OUT_OF_DATE
resolution_budget = (
_resolution_budget
if _resolution_budget is not None
else [MAX_RESOLUTION_WORK_ITEMS]
)
match_budget = (
_match_budget if _match_budget is not None else [MAX_LOCKFILE_MATCH_ATTEMPTS]
)
standalone = _validated_envs is None
package_refs = 0
batch_expansion_budget = (
_batch_expansion_budget
if _batch_expansion_budget is not None
else DocumentExpansionBudget(
items=MAX_LOCKFILE_BATCH_EXPANDED_ITEMS,
strings=MAX_LOCKFILE_BYTES,
)
)
channel_locations_cache = (
_channel_locations_cache if _channel_locations_cache is not None else {}
)
if environment is not None:
config.get_environment(environment)
try:
if isinstance(lockfile_data, CondaLockLoader):
loader = lockfile_data
lockfile_data = loader._data
else:
loader = CondaLockLoader(LOCKFILE_NAME, data=lockfile_data)
lockfile_data = loader._data
except ValueError as exc:
return LockfileStatus(status=_stale, reason=str(exc))
if _validated_envs is None:
declaration_status, validated_envs = loader._validate_manifest(
config, resolution_budget
)
if declaration_status is not None:
return declaration_status
else:
validated_envs = _validated_envs
lock_envs = lockfile_data["environments"]
records_by_url: dict[str, dict[str, Any]] | None = None
env_names = (environment,) if environment is not None else config.environments
for env_name in env_names:
lock_env = lock_envs[env_name]
base_resolved = validated_envs[env_name]
manifest_platforms = base_resolved.platforms or config.platforms
lock_packages = lock_env.get("packages", {})
if _exact_target:
current_lock_platform = current_platform
else:
try:
current_lock_platform = config.resolve_platform_name(
current_platform,
manifest_platforms,
)
except PlatformError:
current_lock_platform = current_platform
platform_refs = lock_packages.get(current_lock_platform)
if platform_refs is None:
continue
if standalone:
if not isinstance(platform_refs, list):
return LockfileStatus(
status=_stale,
reason="Lockfile package references must be a list",
)
package_refs += len(platform_refs)
if package_refs > MAX_LOCKFILE_BATCH_PACKAGE_REFS:
return LockfileStatus(
status=_stale,
reason="Lockfile package references exceed the allowed limit",
)
for ref in platform_refs:
if not isinstance(ref, dict) or set(ref) != {"conda"}:
return LockfileStatus(
status=_stale,
reason=(
f"Environment '{env_name}' package refs must contain "
"exactly one 'conda' entry"
),
)
package_platform = config.platform_subdir(current_lock_platform)
try:
channel_key = (id(lock_env), package_platform)
channel_locations = channel_locations_cache.get(channel_key)
if channel_locations is None:
channel_urls = CondaLockLoader.channel_urls_for_env_data(
lock_env,
package_platform,
)
channel_locations = CondaLockLoader._channel_locations(channel_urls)
channel_locations_cache[channel_key] = channel_locations
if records_by_url is None:
records_by_url = loader._records_index()
remaining_items, remaining_strings = loader._environment_budget(lock_env)
records = CondaLockLoader._package_records_from_refs(
lock_env,
platform_refs,
env_name,
current_lock_platform,
records_by_url,
package_platform=package_platform,
remaining_items=remaining_items,
remaining_strings=remaining_strings,
batch_expansion_budget=batch_expansion_budget,
)
except ValueError as exc:
return LockfileStatus(status=_stale, reason=str(exc))
records_by_name = {}
for record in records:
url = record.url
if not isinstance(url, str) or not url:
return LockfileStatus(
status=_stale,
reason=(
f"Environment '{env_name}' contains an invalid "
f"package ref on '{current_lock_platform}'"
),
)
if not CondaLockLoader._url_matches_locations(url, channel_locations):
return LockfileStatus(
status=_stale,
reason=(
f"Package URL '{redact_url(url)}' in environment "
f"'{env_name}' "
f"on '{current_lock_platform}' is not under a declared "
"channel"
),
)
if record.name in records_by_name:
return LockfileStatus(
status=_stale,
reason=(
f"Environment '{env_name}' contains more than one "
f"'{record.name}' package on '{current_lock_platform}'"
),
)
records_by_name[record.name] = record
from conda.base.context import context as conda_context
from conda.models.match_spec import MatchSpec
from .envs import _build_pypi_specs
env_obj = config.get_environment(env_name)
resolution_budget[0] -= config.resolution_work_items(
env_obj, current_lock_platform
)
if resolution_budget[0] < 0:
return LockfileStatus(
status=_stale,
reason=(
"Workspace environment resolution work exceeds the allowed limit"
),
)
if _exact_target:
target_resolved = replace(
base_resolved,
conda_dependencies=config.merged_conda_dependencies(
env_obj, current_lock_platform
),
pypi_dependencies=config.merged_pypi_dependencies(
env_obj, current_lock_platform
),
system_requirements=config.merged_system_requirements(
env_obj, current_lock_platform
),
)
else:
from .resolver import resolve_environment
target_resolved = resolve_environment(
config,
env_name,
current_lock_platform,
)
requested_specs = [
*target_resolved.conda_dependencies.values(),
*_build_pypi_specs(target_resolved),
]
requested_specs.extend(target_resolved.system_requirement_specs())
try:
dependencies = [
(record, MatchSpec(dependency))
for record in records
for dependency in record.depends
]
constraints = [
(record, MatchSpec(constraint))
for record in records
for constraint in record.constrains
]
except ValueError as exc:
return LockfileStatus(status=_stale, reason=str(exc))
virtual_names = {
spec.name
for spec in (
*requested_specs,
*(spec for _, spec in dependencies),
*(spec for _, spec in constraints),
)
if spec.name and spec.name.startswith("__")
}
candidates = list(records)
if virtual_names:
with (
target_resolved.scoped_virtual_packages(package_platform),
conda_context._override("_subdir", package_platform),
):
candidates.extend(
conda_context.plugin_manager.get_virtual_package_records()
)
candidates_by_name = {record.name: record for record in candidates}
for spec in requested_specs:
dep_name = spec.name
record = candidates_by_name.get(dep_name)
if record is None:
return LockfileStatus(
status=_stale,
reason=(
f"Dependency '{dep_name}' is required by "
f"environment '{env_name}' but not found in "
f"the lockfile for platform "
f"'{current_lock_platform}'"
),
)
if not spec.match(record):
return LockfileStatus(
status=_stale,
reason=(
f"Dependency '{dep_name}' in environment "
f"'{env_name}' requires '{spec}' but locked package "
f"'{record.dist_str()}' on '{current_lock_platform}' "
"does not satisfy it"
),
)
for record, spec in dependencies:
dep_name = spec.get_exact_value("name")
if dep_name is None:
satisfied = False
for candidate in candidates:
if match_budget[0] == 0:
return LockfileStatus(
status=_stale,
reason=(
"Lockfile dependency matching exceeds the allowed limit"
),
)
if _total_match_budget is not None and _total_match_budget[0] == 0:
return LockfileStatus(
status=_stale,
reason=(
"Lockfile dependency matching exceeds the batch limit"
),
)
match_budget[0] -= 1
if _total_match_budget is not None:
_total_match_budget[0] -= 1
if spec.match(candidate):
satisfied = True
break
else:
candidate = candidates_by_name.get(dep_name)
satisfied = candidate is not None and spec.match(candidate)
if not satisfied:
return LockfileStatus(
status=_stale,
reason=(
f"Package '{record.dist_str()}' in environment "
f"'{env_name}' requires '{spec}', which is missing "
f"on '{current_lock_platform}'"
),
)
for record, spec in constraints:
candidate = candidates_by_name.get(spec.name)
if candidate is not None and not spec.match(candidate):
return LockfileStatus(
status=_stale,
reason=(
f"Package '{record.dist_str()}' in environment "
f"'{env_name}' constrains '{spec}', but "
f"'{candidate.dist_str()}' does not satisfy it on "
f"'{current_lock_platform}'"
),
)
return LockfileStatus(status=LockfileStatus.UP_TO_DATE)
[docs]
class CondaLockLoader(EnvironmentSpecBase):
"""Environment specifier + loader for ``conda.lock``.
``conda.lock`` is a derivative of rattler-lock v6 (``pixi.lock``);
this loader shares the rattler-lock v6 conversion helper from
:mod:`conda_lockfiles.rattler_lock.v6` by performing an in-memory
``version: 1 -> 6`` swap before handing the data off. The on-disk
file keeps ``version: 1`` unchanged.
Used by ``conda env create --file conda.lock`` (single platform via
``env``) and by ``conda workspace install`` (multi-platform via
``env_for``).
"""
detection_supported: ClassVar[bool] = True
def __init__(self, path: PathType, *, data: dict[str, Any] | None = None) -> None:
self.path = Path(path).resolve()
self._data_cache = self.redact_data_urls(data) if data is not None else None
self._environment_names_cache: tuple[str, ...] | None = None
self._records_by_url_cache: dict[str, dict[str, Any]] | None = None
self._environment_budgets: dict[int, tuple[int, int]] = {}
def can_handle(self) -> bool:
if self.path.name not in DEFAULT_FILENAMES:
return False
if not self.path.exists():
return False
try:
version = self._data.get("version")
return type(version) is int and version == LOCKFILE_VERSION
except Exception:
return False
@property
def _data(self) -> dict[str, Any]:
if self._data_cache is None:
self._data_cache = self.redact_data_urls(load_lockfile_path(self.path))
return self._data_cache
def _validate_manifest(
self, config: WorkspaceConfig, resolution_budget: list[int]
) -> tuple[LockfileStatus | None, dict[str, ResolvedEnvironment]]:
"""Check lock metadata once and retain each environment's platform list."""
stale = LockfileStatus.OUT_OF_DATE
data = self._data
version = data.get("version")
if type(version) is not int:
return (
LockfileStatus(
status=stale, reason="Lockfile version must be an integer"
),
{},
)
if version != LOCKFILE_VERSION:
return (
LockfileStatus(
status=stale,
reason=(
f"Lockfile version {version!r} "
f"does not match expected version {LOCKFILE_VERSION}"
),
),
{},
)
from .resolver import resolve_environment
lock_envs = data.get("environments", {})
extra_envs = sorted(set(lock_envs) - set(config.environments))
if extra_envs:
names = ", ".join(f"'{name}'" for name in extra_envs)
return (
LockfileStatus(
status=stale,
reason=(
f"Environment{'' if len(extra_envs) == 1 else 's'} {names} "
f"{'is' if len(extra_envs) == 1 else 'are'} present in the "
"lockfile but not declared in the manifest"
),
),
{},
)
resolved_envs: dict[str, ResolvedEnvironment] = {}
platform_sets: dict[int, set[str]] = {}
declaration_items = 0
target_count = 0
for env_name, env_obj in config.environments.items():
if env_name not in lock_envs:
return (
LockfileStatus(
status=stale,
reason=(
f"Environment '{env_name}' is declared in the manifest "
"but missing from the lockfile"
),
),
{},
)
resolution_budget[0] -= config.resolution_work_items(env_obj)
if resolution_budget[0] < 0:
return (
LockfileStatus(
status=stale,
reason=(
"Workspace environment resolution work "
"exceeds the allowed limit"
),
),
{},
)
lock_env = lock_envs[env_name]
manifest_channels = _manifest_channel_entries(config, env_obj)
manifest_urls = _normalize_channel_urls(
entry["url"] for entry in manifest_channels
)
lock_channel_entries = lock_env.get("channels", [])
declaration_items += len(manifest_channels) + len(lock_channel_entries)
if declaration_items > MAX_LOCKFILE_DECLARATION_ITEMS:
return (
LockfileStatus(
status=stale,
reason="Lockfile declaration work exceeds the allowed limit",
),
{},
)
lock_urls = _normalize_channel_urls(
entry.get("url", "") for entry in lock_channel_entries
)
if manifest_urls != lock_urls:
return (
LockfileStatus(
status=stale,
reason=(
f"Channel mismatch for environment '{env_name}': "
"manifest declares "
f"{[redact_channel_name(url) for url in manifest_urls]} "
"but lockfile has "
f"{[redact_channel_name(url) for url in lock_urls]}"
),
),
{},
)
resolved = resolve_environment(config, env_name)
declaration_items += (
len(resolved.conda_dependencies)
+ len(resolved.pypi_dependencies)
+ len(resolved.system_requirements)
+ len(resolved.channels)
+ len(resolved.platform_subdirs)
+ len(resolved.activation_scripts)
+ len(resolved.activation_env)
)
if declaration_items > MAX_LOCKFILE_DECLARATION_ITEMS:
return (
LockfileStatus(
status=stale,
reason="Lockfile declaration work exceeds the allowed limit",
),
{},
)
manifest_platforms = resolved.platforms or config.platforms
target_count += len(manifest_platforms) or 1
declaration_items += len(manifest_platforms)
if target_count > MAX_LOCKFILE_BATCH_TARGETS:
return (
LockfileStatus(
status=stale,
reason="Lockfile status target count exceeds the allowed limit",
),
{},
)
if declaration_items > MAX_LOCKFILE_DECLARATION_ITEMS:
return (
LockfileStatus(
status=stale,
reason="Lockfile declaration work exceeds the allowed limit",
),
{},
)
lock_packages = lock_env.get("packages", {})
lock_platforms = platform_sets.get(id(lock_packages))
if lock_platforms is None:
declaration_items += len(lock_packages)
if declaration_items > MAX_LOCKFILE_DECLARATION_ITEMS:
return (
LockfileStatus(
status=stale,
reason=(
"Lockfile declaration work exceeds the allowed limit"
),
),
{},
)
lock_platforms = set(lock_packages)
platform_sets[id(lock_packages)] = lock_platforms
for platform in manifest_platforms:
if platform not in lock_platforms:
return (
LockfileStatus(
status=stale,
reason=(
f"Platform '{platform}' is declared in the "
f"manifest but missing from environment "
f"'{env_name}' in the lockfile"
),
),
{},
)
resolved_envs[env_name] = resolved
return None, resolved_envs
def check_targets(
self,
config: WorkspaceConfig,
targets: Iterable[tuple[str, str]] | None = None,
*,
match_budget_scope: Literal["target", "platform"],
default_platform: str | None = None,
) -> LockfileStatus:
"""Check exact environment/platform targets after one manifest validation.
Platform management retains its prior per-target wildcard budget.
Lockfile rendering retains a shared budget for each platform. Both
modes also enforce one operation-wide wildcard comparison limit.
Omit *targets* to check every declared target, using *default_platform*
(or conda's current subdir) for environments without platform lists.
"""
from conda.base.context import context as conda_context
from .resolver import MAX_RESOLUTION_WORK_ITEMS
if match_budget_scope not in ("target", "platform"):
raise ValueError("Invalid lockfile match budget scope")
resolution_budget = [MAX_RESOLUTION_WORK_ITEMS]
declaration_status, resolved_envs = self._validate_manifest(
config, resolution_budget
)
if declaration_status is not None:
return declaration_status
if targets is None:
targets = (
(name, platform)
for name, resolved in resolved_envs.items()
for platform in (
resolved.platforms or [default_platform or conda_context.subdir]
)
)
declared = {
name: set(resolved.platforms or config.platforms)
for name, resolved in resolved_envs.items()
}
platform_budgets: dict[str, list[int]] = {}
total_budget = [MAX_LOCKFILE_BATCH_MATCH_ATTEMPTS]
expansion_budget = DocumentExpansionBudget(
items=MAX_LOCKFILE_BATCH_EXPANDED_ITEMS,
strings=MAX_LOCKFILE_BYTES,
)
channel_locations_cache: dict[tuple[int, str], ChannelLocations] = {}
seen: set[tuple[str, str]] = set()
package_refs = 0
for inspected, (name, platform) in enumerate(targets, 1):
if inspected > MAX_LOCKFILE_BATCH_TARGETS:
return LockfileStatus(
status=LockfileStatus.OUT_OF_DATE,
reason="Lockfile status target count exceeds the allowed limit",
)
if (name, platform) in seen:
continue
seen.add((name, platform))
config.get_environment(name)
if declared[name] and platform not in declared[name]:
return LockfileStatus(
status=LockfileStatus.OUT_OF_DATE,
reason=(
f"Platform '{platform}' is not declared for "
f"environment '{name}'"
),
)
refs = self._data["environments"][name].get("packages", {}).get(platform)
if refs is not None:
if not isinstance(refs, list):
return LockfileStatus(
status=LockfileStatus.OUT_OF_DATE,
reason="Lockfile package references must be a list",
)
package_refs += len(refs)
if package_refs > MAX_LOCKFILE_BATCH_PACKAGE_REFS:
return LockfileStatus(
status=LockfileStatus.OUT_OF_DATE,
reason="Lockfile package references exceed the allowed limit",
)
match_budget = (
[MAX_LOCKFILE_MATCH_ATTEMPTS]
if match_budget_scope == "target"
else platform_budgets.setdefault(
platform, [MAX_LOCKFILE_MATCH_ATTEMPTS]
)
)
status = check_lockfile_satisfiability(
config,
self,
platform,
environment=name,
_validated_envs=resolved_envs,
_match_budget=match_budget,
_total_match_budget=total_budget,
_batch_expansion_budget=expansion_budget,
_channel_locations_cache=channel_locations_cache,
_resolution_budget=resolution_budget,
_exact_target=True,
)
if status.status != LockfileStatus.UP_TO_DATE:
return status
return LockfileStatus(status=LockfileStatus.UP_TO_DATE)
@property
def available_platforms(self) -> tuple[str, ...]:
"""Platforms declared in this lockfile's default environment."""
return self.platforms_for()
@property
def available_environments(self) -> tuple[str, ...]:
"""Return the environment names declared in this lockfile."""
if self._environment_names_cache is not None:
return self._environment_names_cache
data = self._data
version = data.get("version")
if type(version) is not int:
raise ValueError("Lockfile version must be an integer")
if version != LOCKFILE_VERSION:
raise ValueError(
f"Unsupported {LOCKFILE_NAME} version: {version!r} "
f"(expected {LOCKFILE_VERSION})"
)
environments = data.get("environments")
if not isinstance(environments, dict) or not all(
isinstance(name, str) and name for name in environments
):
raise ValueError("Lockfile environments must be a mapping of names")
self._environment_names_cache = tuple(sorted(environments))
return self._environment_names_cache
[docs]
def select(self, selections: Mapping[str, Iterable[str]]) -> dict[str, Any]:
"""Copy selected solutions and their source metadata without solving."""
from conda.base.context import context
if not isinstance(selections, Mapping) or not selections:
raise ValueError("Select at least one lockfile environment and target")
result = self.redact_data_urls(self._data)
selected_environments: dict[str, Any] = {}
selected_urls: set[str] = set()
for name, requested in selections.items():
env_data = deepcopy(self._env_data(name))
if isinstance(requested, (str, bytes)):
raise ValueError("Lockfile targets must be a collection of names")
try:
targets = tuple(requested)
except TypeError as exc:
raise ValueError(
"Lockfile targets must be a collection of names"
) from exc
if not targets or not all(isinstance(target, str) for target in targets):
raise ValueError("Select at least one named lockfile target")
for target in targets:
if target not in env_data["packages"]:
raise ValueError(
f"Environment {name!r} does not include platform {target!r}"
)
records = self.package_records_for_env_data(result, name, target)
subdirs = {
record.subdir for record in records if record.subdir != "noarch"
}
if len(subdirs) > 1 or not subdirs.issubset(context.known_subdirs):
raise ValueError(
f"Environment {name!r} target {target!r} contains "
"packages without one supported conda subdir"
)
self.validate_env_for_conversion(
result,
name,
target,
package_platform=next(iter(subdirs), target),
)
selected_urls.update(
ref["conda"] for ref in env_data["packages"][target]
)
env_data["packages"] = {
target: refs
for target, refs in env_data["packages"].items()
if target in targets
}
selected_environments[name] = env_data
result["environments"] = selected_environments
result["packages"] = [
record
for url, record in self.package_records_by_url_from_data(result).items()
if url in selected_urls
]
return result
[docs]
def env_for(
self,
platform: str,
name: str = "default",
*,
package_platform: str | None = None,
metadata_only: bool = False,
) -> Environment:
"""Return the conda ``Environment`` for *platform* and *name*.
Raises ``PlatformMismatchError`` if *platform* is not in the
lockfile or *name* does not identify a declared environment.
*package_platform* supplies the backing conda subdir when
*platform* is a rich workspace platform name. The returned environment
keeps the backing subdir as its conda platform and records the logical
lock key separately for round-trip serialization.
*metadata_only* reconstructs exact records from the lockfile without
accessing the package cache.
"""
return self.envs_for(
{platform: package_platform},
name,
metadata_only=metadata_only,
)[0]
def envs_for(
self,
platforms: Mapping[str, str | None],
name: str = "default",
*,
metadata_only: bool = False,
) -> list[Environment]:
"""Convert selected targets after copying and indexing the lockfile once."""
self._env_data(name)
payload = self.redact_data_urls(self._data)
env_data = payload["environments"][name]
available = tuple(sorted(env_data["packages"]))
for platform in platforms:
if platform not in env_data["packages"]:
from conda.exceptions import PlatformMismatchError
raise PlatformMismatchError(
incompatible=[(str(self.path), available)],
subdir=platform,
)
if not metadata_only:
DocumentExpansionBudget(
items=MAX_LOCKFILE_ITEMS,
strings=MAX_LOCKFILE_BYTES,
).consume(payload, label="Lockfile")
try:
records_by_url = self.package_records_by_url_from_data(payload)
except ValueError as exc:
raise LockfileIntegrityError(self.path, str(exc)) from exc
envs: list[Environment] = []
for platform, package_platform in platforms.items():
conversion_platform = package_platform or platform
refs = env_data["packages"][platform]
selected_urls = dict.fromkeys(
ref["conda"]
for ref in refs
if isinstance(ref, dict) and isinstance(ref.get("conda"), str)
)
selected_records = [
records_by_url[url] for url in selected_urls if url in records_by_url
]
selected_env_data = {
"channels": env_data["channels"],
"packages": {platform: refs},
}
selected_payload = {
"version": LOCKFILE_VERSION,
"environments": {name: selected_env_data},
"packages": selected_records,
}
self.validate_env_for_conversion(
selected_payload,
name,
platform,
package_platform=conversion_platform,
records_by_url=records_by_url,
)
if metadata_only:
from conda.models.channel import Channel
from conda.models.environment import Environment, EnvironmentConfig
from conda_lockfiles.rattler_lock.v6 import RattlerLockV6Environment
records = self.package_records_for_env_data(
selected_payload,
name,
platform,
package_platform=conversion_platform,
)
lock_environment = RattlerLockV6Environment.model_validate(
selected_env_data
)
env = Environment(
name=name,
platform=conversion_platform,
config=EnvironmentConfig(
channels=tuple(
Channel(channel.url).canonical_name
for channel in lock_environment.channels
)
),
explicit_packages=records,
)
else:
from conda_lockfiles.rattler_lock.v6 import (
rattler_lock_v6_to_conda_env,
)
from ._lockfile_compat import WorkspaceLock
# The shared rattler model requires a default environment,
# while conda.lock may contain only a named environment.
conversion_payload = {
"version": 6,
"environments": {
"default": {
"channels": selected_env_data["channels"],
"packages": {conversion_platform: refs},
}
},
"packages": selected_records,
}
DocumentExpansionBudget(
items=MAX_LOCKFILE_ITEMS,
strings=MAX_LOCKFILE_BYTES,
).consume(conversion_payload, label="Lockfile")
lockfile_model = WorkspaceLock.model_validate(conversion_payload)
env = rattler_lock_v6_to_conda_env(
lockfile_model,
name="default",
platform=conversion_platform,
)
env.name = name
if conversion_platform != platform:
setattr(env, "lock_platform", platform)
envs.append(env)
return envs
def validate_env_for_conversion(
self,
data: dict[str, Any],
name: str,
platform: str,
*,
package_platform: str | None = None,
records_by_url: Mapping[str, dict[str, Any]] | None = None,
) -> None:
"""Validate one environment target before rattler conversion."""
env_data = data["environments"][name]
refs = env_data.get("packages", {}).get(platform, ())
channel_urls = self.channel_urls_for_env_data(
env_data,
package_platform or platform,
path=self.path,
)
channel_locations = self._channel_locations(channel_urls)
if records_by_url is None:
try:
records_by_url = self.package_records_by_url_from_data(data)
except ValueError as exc:
raise LockfileIntegrityError(self.path, str(exc)) from exc
for ref in refs:
if not isinstance(ref, dict):
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid package ref",
)
if set(ref) != {"conda"}:
if ref:
raise LockfileIntegrityError(
self.path,
"external package refs cannot be verified from conda.lock. "
"Declare them in the workspace manifest, regenerate the "
"lockfile, then use 'conda workspace install'",
)
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid package ref",
)
url = ref["conda"]
if not isinstance(url, str) or not url:
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid conda package ref",
)
if not self._url_matches_locations(url, channel_locations):
raise LockfileIntegrityError(
self.path,
f"package URL {redact_url(url)!r} is not under any channel "
f"declared for environment {name!r}",
)
record = records_by_url.get(url)
if record is None:
raise LockfileIntegrityError(
self.path,
f"package URL {redact_url(url)!r} has no top-level package record",
)
self.digest_fragment_for_record(record, url, path=self.path)
@property
def env(self) -> Environment:
"""Return the default environment for the current subdir.
Kept for backwards compatibility with ``conda env create --file
conda.lock``; delegates to :meth:`env_for`.
"""
from conda.base.context import context
return self.env_for(context.subdir)
def _env_data(self, name: str = "default") -> dict[str, Any]:
available = self.available_environments
if name not in available:
raise ValueError(
f"Environment {name!r} not found in lockfile. "
f"Available environments: {dashlist(available)}"
)
env_data = self._data["environments"][name]
if not isinstance(env_data, dict):
raise ValueError(f"Lockfile environment {name!r} must be a mapping")
packages = env_data.get("packages")
if not isinstance(packages, dict) or not all(
isinstance(target, str) and target and isinstance(refs, list)
for target, refs in packages.items()
):
raise ValueError(
f"Lockfile environment {name!r} has invalid target packages"
)
channels = env_data.get("channels")
if not isinstance(channels, list) or not all(
isinstance(channel, dict)
and isinstance(channel.get("url"), str)
and channel["url"]
for channel in channels
):
raise ValueError(f"Lockfile environment {name!r} has invalid channels")
return env_data
def explicit_package_specs_for(
self,
platform: str,
name: str = "default",
*,
package_platform: str | None = None,
) -> list[str]:
"""Return hash-bearing explicit conda specs for *name* on *platform*."""
env_data = self._env_data(name)
platform_refs = env_data.get("packages", {}).get(platform)
if platform_refs is None:
raise ValueError(
f"Environment {name!r} does not include packages for {platform!r}"
)
records_by_url = self._records_index()
channel_urls = self.channel_urls_for(
env_data,
package_platform or platform,
)
channel_locations = self._channel_locations(channel_urls)
explicit_specs: list[str] = []
for ref in platform_refs:
if not isinstance(ref, dict):
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid package ref",
)
if set(ref) != {"conda"}:
if ref:
raise LockfileIntegrityError(
self.path,
"external package refs cannot be installed exactly from "
"conda.lock. Declare them in the workspace manifest and "
"regenerate the lockfile",
)
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid package ref",
)
url = ref["conda"]
if not isinstance(url, str) or not url:
raise LockfileIntegrityError(
self.path,
f"environment {name!r} has an invalid conda package ref",
)
if not self._url_matches_locations(url, channel_locations):
raise LockfileIntegrityError(
self.path,
f"package URL {redact_url(url)!r} is not under any channel "
"declared "
f"for environment {name!r}",
)
record = records_by_url.get(url)
if record is None:
raise LockfileIntegrityError(
self.path,
f"package URL {redact_url(url)!r} has no top-level package record",
)
digest = self.digest_fragment_for(record, url)
explicit_specs.append(f"{redact_url(url)}#{digest}")
return explicit_specs
def package_records_by_url(self) -> dict[str, dict[str, Any]]:
"""Return top-level conda package records keyed by their exact URL."""
return dict(self._records_index())
def _records_index(self) -> dict[str, dict[str, Any]]:
"""Cache the package URL index for repeated target selection."""
if self._records_by_url_cache is None:
self._records_by_url_cache = self.package_records_by_url_from_data(
self._data
)
return self._records_by_url_cache
def _environment_budget(self, env_data: dict[str, Any]) -> tuple[int, int]:
"""Cache resources left after accounting for an environment's refs."""
identity = id(env_data)
if identity not in self._environment_budgets:
budget = DocumentExpansionBudget(
items=MAX_LOCKFILE_ITEMS,
strings=MAX_LOCKFILE_BYTES,
)
budget.consume({"environment": env_data}, label="Lockfile")
self._environment_budgets[identity] = budget.items, budget.strings
return self._environment_budgets[identity]
@staticmethod
def package_records_by_url_from_data(
data: dict[str, Any],
) -> dict[str, dict[str, Any]]:
"""Return top-level conda package records from *data* keyed by URL."""
records_by_url: dict[str, dict[str, Any]] = {}
equal_value_pairs: dict[tuple[int, int], tuple[object, object]] = {}
validated_paths: set[str] = set()
packages = data.get("packages", [])
if not isinstance(packages, list):
raise ValueError("Lockfile package records must be a list")
for record in packages:
if not isinstance(record, dict):
continue
url = record.get("conda") or record.get("url") or record.get("pypi")
if isinstance(url, str) and url:
site_packages_path = record.get("python_site_packages_path")
if site_packages_path is not None:
try:
if not isinstance(site_packages_path, str):
raise ValueError
if site_packages_path not in validated_paths:
parse_relative_posix_path(site_packages_path)
validated_paths.add(site_packages_path)
except ValueError:
raise ValueError(
"Lockfile python_site_packages_path must be a relative "
"POSIX path"
) from None
existing = records_by_url.get(url)
if existing is not None and not document_values_equal(
existing,
record,
equal_value_pairs=equal_value_pairs,
):
raise ValueError(
f"Conflicting package metadata for URL {redact_url(url)!r}"
)
records_by_url.setdefault(url, record)
return records_by_url
@staticmethod
def merge_package_records(
existing: dict[str, Any],
update: dict[str, Any],
url: str,
) -> dict[str, Any]:
"""Merge repodata patches with installed-prefix metadata.
Those sources can disagree on non-integrity metadata for the same
package URL. Only conflicting hashes identify an integrity conflict.
"""
conflicts = {
key
for key in (existing.keys() & update.keys()) & {"sha256", "md5"}
if existing[key] != update[key]
}
if conflicts:
raise ValueError(
"Conflicting package metadata for URL"
f" {redact_url(url)!r}: {', '.join(sorted(conflicts))}"
)
source_keys = {"conda", "pypi", "url"}
merged = {
**{key: value for key, value in existing.items() if key not in source_keys},
**update,
}
for key in ("depends", "constrains", "features", "track_features"):
if key in update and not update[key]:
merged.pop(key, None)
return merged
@classmethod
def package_records_for_env_data(
cls,
data: dict[str, Any],
name: str,
platform: str,
*,
package_platform: str | None = None,
) -> list[PackageRecord]:
"""Reconstruct package records for one lockfile environment target.
This path uses the metadata already embedded in ``conda.lock``. It
deliberately avoids :meth:`env_for`, whose generic rattler-lock
conversion may fetch package archives to fill missing metadata.
"""
return cls(LOCKFILE_NAME, data=data).package_records_for(
name,
platform,
package_platform=package_platform,
)
def package_records_for(
self,
name: str,
platform: str,
*,
package_platform: str | None = None,
) -> list[PackageRecord]:
"""Reconstruct one target using the sanitized lockfile and cached index."""
data = self._data
environments = data.get("environments", {})
env_data = environments.get(name)
if not isinstance(env_data, dict):
raise ValueError(f"Environment {name!r} is missing from the lockfile")
refs = env_data.get("packages", {}).get(platform)
if refs is None:
raise ValueError(
f"Environment {name!r} does not include platform {platform!r}"
)
records_by_url = self._records_index()
remaining_items, remaining_strings = self._environment_budget(env_data)
return self._package_records_from_refs(
env_data,
refs,
name,
platform,
records_by_url,
package_platform=package_platform,
remaining_items=remaining_items,
remaining_strings=remaining_strings,
)
@classmethod
def _package_records_from_refs(
cls,
env_data: dict[str, Any],
refs: list[dict[str, Any]],
name: str,
platform: str,
records_by_url: Mapping[str, dict[str, Any]],
*,
package_platform: str | None = None,
remaining_items: int,
remaining_strings: int,
batch_expansion_budget: DocumentExpansionBudget | None = None,
) -> list[PackageRecord]:
"""Reconstruct one target from sanitized data and a shared index.
Callers account for expanded environment refs once. This method charges
selected package records against the remaining expansion budget.
"""
from conda.base.context import context
from conda.models.records import PackageRecord
selected_metadata: list[dict[str, Any]] = []
for ref in refs:
if isinstance(ref, dict):
url = ref.get("conda")
if isinstance(url, str) and url in records_by_url:
selected_metadata.append(records_by_url[url])
if batch_expansion_budget is not None:
batch_expansion_budget.consume(
{"packages": selected_metadata},
label="Lockfile batch",
)
DocumentExpansionBudget(
items=remaining_items,
strings=remaining_strings,
).consume(
{"packages": selected_metadata},
label="Lockfile",
)
records: list[PackageRecord] = []
for ref in refs:
if not isinstance(ref, dict) or "conda" not in ref:
continue
url = ref["conda"]
if not isinstance(url, str) or not url:
raise ValueError(
f"Environment {name!r} has an invalid package reference"
)
metadata = records_by_url.get(url)
if metadata is None:
raise ValueError(
f"Package URL {redact_url(url)!r} has no top-level record"
)
if "pypi" in metadata or (
"conda" in metadata
and "url" in metadata
and metadata["conda"] != metadata["url"]
):
raise ValueError(
f"Package URL {redact_url(url)!r} has inconsistent package sources"
)
cls.digest_fragment_for_record(metadata, url)
package_url = redact_url(url)
dist = Dist(package_url)
identity = {
"name": dist.name,
"version": dist.version,
"build": dist.build_string,
"subdir": dist.subdir,
"fn": dist.to_filename(),
}
for key, value in identity.items():
if metadata.get(key) is not None and metadata[key] != value:
raise ValueError(
f"Package URL {package_url!r} disagrees with its {key} metadata"
)
if (
dist.subdir is not None
and (package_platform or platform) in context.known_subdirs
and dist.subdir not in {"noarch", package_platform or platform}
):
raise ValueError(
f"Package URL {package_url!r} does not match target {platform!r}"
)
records.append(
PackageRecord.from_objects(
metadata,
name=dist.name,
version=dist.version,
build=dist.build_string,
build_number=metadata.get("build_number", dist.build_number),
channel=dist.channel,
subdir=dist.subdir or package_platform or platform,
fn=dist.to_filename(),
url=package_url,
)
)
return records
@classmethod
def seed_prefix_from_data(
cls,
data: dict[str, Any] | CondaLockLoader,
name: str,
platform: str,
prefix: Path,
requested_specs: Iterable[MatchSpec],
*,
package_platform: str | None = None,
) -> list[PackageRecord]:
"""Create a metadata-only prefix from one lockfile target."""
from conda.core.prefix_data import PrefixData
from conda.history import History
from conda.models.records import PrefixRecord
loader = (
data if isinstance(data, CondaLockLoader) else cls(LOCKFILE_NAME, data=data)
)
records = loader.package_records_for(
name, platform, package_platform=package_platform
)
(prefix / "conda-meta").mkdir(parents=True)
prefix_data = PrefixData(str(prefix))
for record in records:
prefix_data.insert(PrefixRecord.from_objects(record))
history = History(str(prefix))
history.write_changes(set(), {record.dist_str() for record in records})
history.write_specs(update_specs=tuple(requested_specs))
return records
@staticmethod
def redact_data_urls(data: dict[str, Any]) -> dict[str, Any]:
"""Return lockfile data with every serialized URL credential-free."""
result = plain_document(
data,
maximum_items=MAX_LOCKFILE_ITEMS,
maximum_collection_items=MAX_LOCKFILE_COLLECTION_ITEMS,
string_budget=DocumentStringBudget(MAX_LOCKFILE_BYTES),
)
environments = result.get("environments")
normalized_channels: set[int] = set()
normalized_urls: dict[int, tuple[str, str]] = {}
if isinstance(environments, dict):
for environment in environments.values():
if not isinstance(environment, dict):
continue
channels = environment.get("channels")
if (
isinstance(channels, list)
and id(channels) not in normalized_channels
):
normalized_channels.add(id(channels))
for index, entry in enumerate(channels):
if isinstance(entry, str):
url = entry
elif isinstance(entry, dict):
if id(entry) in normalized_channels:
continue
normalized_channels.add(id(entry))
channel_entry = cast("dict[str, Any]", entry)
url = channel_entry.get("url")
else:
continue
if isinstance(url, str):
if id(url) not in normalized_urls:
normalized_urls[id(url)] = (
url,
redact_channel_name(url),
)
normalized = normalized_urls[id(url)][1]
if isinstance(entry, str):
channels[index] = {"url": normalized}
else:
cast("dict[str, Any]", entry)["url"] = normalized
visited: set[int] = set()
checked_keys: set[int] = set()
string_budget = DocumentStringBudget(MAX_LOCKFILE_BYTES)
# Retain source strings while replacing values so IDs cannot be reused.
redacted_urls: dict[int, tuple[str, str]] = {}
values: list[dict[str, Any] | list[Any]] = [result]
while values:
container = values.pop()
if id(container) in visited:
continue
visited.add(id(container))
if isinstance(container, dict):
for key, value in list(container.items()):
string_budget.consume(key, label="Lockfile data")
if id(key) not in checked_keys:
if has_url_credentials(key):
raise ValueError(
"Lockfile contains credentials in a mapping key"
)
checked_keys.add(id(key))
if isinstance(value, str):
string_budget.consume(value, label="Lockfile data")
if id(value) not in redacted_urls:
redacted_urls[id(value)] = (value, redact_url(value))
container[key] = redacted_urls[id(value)][1]
elif isinstance(value, (dict, list)):
values.append(value)
else:
for index, value in enumerate(container):
if isinstance(value, str):
string_budget.consume(value, label="Lockfile data")
if id(value) not in redacted_urls:
redacted_urls[id(value)] = (value, redact_url(value))
container[index] = redacted_urls[id(value)][1]
elif isinstance(value, (dict, list)):
values.append(value)
if has_url_credentials_in_data(result):
raise ValueError("Lockfile contains credentials in structured metadata")
return result
@classmethod
def replace_solutions(
cls,
baseline: dict[str, Any],
envs: Iterable[_SolvedEnvironment],
*,
platforms_by_environment: Mapping[str, Sequence[str]] | None = None,
) -> dict[str, Any]:
"""Replace solutions and canonicalize referenced package records.
*platforms_by_environment* reconciles platform declarations while
retaining existing solutions and their metadata. New targets may be
added, removed targets are discarded, and shared package hashes must
still agree with newly solved records.
"""
result = cls.redact_data_urls(baseline)
retained_urls: set[str] = set()
if platforms_by_environment is not None:
environments = result.get("environments", {})
for name in platforms_by_environment:
if name not in environments:
raise ValueError(
f"Environment {name!r} is missing from the lockfile"
)
result["environments"] = {
name: environment
for name, environment in environments.items()
if name in platforms_by_environment
}
for name, environment in result["environments"].items():
environment["packages"] = {
platform: refs
for platform, refs in environment.get("packages", {}).items()
if platform in platforms_by_environment[name]
}
for refs in environment["packages"].values():
for ref in refs:
url = ref.get("conda") or ref.get("url") or ref.get("pypi")
if url:
retained_urls.add(url)
updates = cls.compose(envs)
for name, update in updates["environments"].items():
if name not in result.get("environments", {}):
raise ValueError(f"Environment {name!r} is missing from the lockfile")
result_env = result["environments"][name]
if platforms_by_environment is None:
result_env["channels"] = update["channels"]
for platform, refs in update["packages"].items():
allowed = (
result_env.get("packages", {})
if platforms_by_environment is None
else platforms_by_environment[name]
)
if platform not in allowed:
raise ValueError(
f"Environment {name!r} does not include platform {platform!r}"
)
result_env["packages"][platform] = refs
packages_by_url = cls.package_records_by_url_from_data(result)
update_packages_by_url = cls.package_records_by_url_from_data(updates)
for url, update_record in update_packages_by_url.items():
existing_record = packages_by_url.get(url)
if existing_record is None:
packages_by_url[url] = update_record
continue
merged = cls.merge_package_records(
existing_record,
update_record,
url,
)
if url not in retained_urls:
packages_by_url[url] = merged
packages: list[dict[str, Any]] = []
seen_urls: set[str] = set()
for environment in result.get("environments", {}).values():
for platform in sorted(environment.get("packages", {})):
for ref in environment["packages"][platform]:
url = ref.get("conda") or ref.get("url") or ref.get("pypi")
if not url or url in seen_urls:
continue
record = packages_by_url.get(url)
if record is not None:
packages.append(record)
seen_urls.add(url)
result["packages"] = packages
return result
def channel_urls_for(
self,
env_data: dict[str, Any],
platform: str,
) -> tuple[str, ...]:
"""Return concrete package-containing channel URLs for *platform*."""
return self.channel_urls_for_env_data(env_data, platform, path=self.path)
@staticmethod
def channel_urls_for_env_data(
env_data: dict[str, Any],
platform: str,
*,
path: Path | None = None,
) -> tuple[str, ...]:
"""Return concrete package-containing channel URLs for *platform*."""
from conda.models.channel import Channel
urls: set[str] = set()
seen_raw_urls: set[str] = set()
for entry in env_data.get("channels", []) or []:
if not isinstance(entry, dict):
continue
raw_url = entry.get("url")
if not isinstance(raw_url, str) or not raw_url:
continue
if raw_url in seen_raw_urls:
continue
try:
channel = Channel(raw_url)
for with_credentials in (False, True):
urls.update(
channel.urls(
with_credentials=with_credentials,
subdirs=(platform, "noarch"),
)
)
seen_raw_urls.add(raw_url)
except Exception:
reason = (
"environment channel"
f" {redact_channel_name(raw_url)!r} cannot be resolved"
)
if path is None:
raise ValueError(reason) from None
raise LockfileIntegrityError(path, reason) from None
return tuple(sorted(url.rstrip("/") for url in urls))
@staticmethod
def url_location(
url: str,
) -> UrlLocation | None:
"""Return a normalized URL origin and traversal-safe path segments."""
try:
parsed = urlsplit(url)
scheme = parsed.scheme.lower()
hostname = parsed.hostname or ""
if not scheme or (scheme != "file" and not hostname):
return None
hostname = hostname.encode("idna").decode("ascii").lower()
port = parsed.port
except (UnicodeError, ValueError):
return None
if port is None:
port = {"http": 80, "https": 443}.get(scheme)
origin = (
scheme,
hostname,
port,
unquote(parsed.username) if parsed.username is not None else None,
unquote(parsed.password) if parsed.password is not None else None,
)
segments: list[str] = []
for raw_segment in parsed.path.split("/"):
if not raw_segment:
continue
segment = unquote(raw_segment)
if (
segment in {".", ".."}
or "/" in segment
or "\\" in segment
or "\0" in segment
or unquote(segment) != segment
):
return None
segments.append(segment)
return origin, tuple(segments)
@classmethod
def url_matches_channel(cls, url: str, channel_urls: tuple[str, ...]) -> bool:
"""Return whether *url* is a traversal-safe child of a declared channel."""
return cls._url_matches_locations(url, cls._channel_locations(channel_urls))
@classmethod
def _channel_locations(cls, channel_urls: Iterable[str]) -> ChannelLocations:
"""Index declared channel paths by origin and decoded segments."""
locations: ChannelLocations = {}
for channel_url in channel_urls:
location = cls.url_location(channel_url)
if location is not None:
origin, path = location
node = locations.setdefault(origin, {})
for segment in path:
node = node.setdefault(segment, {})
# Empty segments are omitted by url_location, so this marks
# a complete channel path without colliding with a URL segment.
node[""] = {}
return locations
@classmethod
def _url_matches_locations(
cls,
url: str,
channel_locations: ChannelLocations,
) -> bool:
"""Follow a package path through declared channels for its URL origin."""
package_location = cls.url_location(url)
if package_location is None:
return False
package_origin, package_path = package_location
node = channel_locations.get(package_origin)
if node is None:
return False
for segment in package_path:
if "" in node:
return True
node = node.get(segment)
if node is None:
return False
return False
def digest_fragment_for(self, record: dict[str, Any], url: str) -> str:
"""Return the conda explicit-file digest fragment for *record*."""
return self.digest_fragment_for_record(record, url, path=self.path)
@classmethod
def digest_fragment_for_record(
cls,
record: dict[str, Any],
url: str,
*,
path: Path | None = None,
) -> str:
"""Return the conda explicit-file digest fragment for *record*."""
sha256 = record.get("sha256")
if sha256 is not None:
if cls.is_hex_digest(sha256, 64):
return f"sha256:{sha256.lower()}"
reason = f"package URL {redact_url(url)!r} has an invalid sha256 digest"
if path is None:
raise ValueError(reason)
raise LockfileIntegrityError(path, reason)
md5 = record.get("md5")
if md5 is not None:
if cls.is_hex_digest(md5, 32):
return md5.lower()
reason = f"package URL {redact_url(url)!r} has an invalid md5 digest"
if path is None:
raise ValueError(reason)
raise LockfileIntegrityError(path, reason)
reason = f"package URL {redact_url(url)!r} is missing a sha256 or md5 digest"
if path is None:
raise ValueError(reason)
raise LockfileIntegrityError(path, reason)
@staticmethod
def is_hex_digest(value: object, length: int) -> bool:
"""Return whether *value* is a hex digest with *length* characters."""
if not isinstance(value, str) or len(value) != length:
return False
try:
int(value, 16)
except ValueError:
return False
return True
@classmethod
def compose(
cls,
envs: Iterable[Environment | _SolvedEnvironment],
) -> dict[str, Any]:
"""Compose ``Environment`` objects into a ``conda.lock`` dict.
The write-side companion to :meth:`env_for`: same loader class
owns both directions. Returned dict has the canonical
``version`` / ``environments`` / ``packages`` shape that
:func:`.export.multiplatform_export` (our
``conda-workspaces-lock-v1`` plugin callable) then hands to
conda's YAML dumper. Exposed as a public classmethod because
callers that want to inspect, merge, or hand off to a
different serialiser can reuse the same composition logic
without re-implementing it.
"""
from conda.models.environment import Environment, EnvironmentConfig
from conda_lockfiles.validate_urls import validate_urls
from ._lockfile_compat import WorkspaceLockPackage
packages: list[dict[str, Any]] = []
environments: dict[str, dict[str, Any]] = {}
for env in envs:
# ruamel.yaml dispatches representers by exact type on dict
# keys, so any ``str`` subclass reaching this point (e.g. a
# leaked ``tomlkit.items.String``) raises ``TypeError:
# Object of type ... is not YAML serializable``. Workspace
# parsers unwrap tomlkit docs at load time; this is the
# last-line guard for callers that build ``Environment``
# objects through other paths (``conda export`` plugin,
# tests, third parties).
env_name = str(env.name or "default")
platform = str(getattr(env, "lock_platform", env.platform))
package_platform = str(getattr(env, "package_platform", env.platform))
validation_env = env
if str(env.platform) != package_platform:
validation_env = Environment(
name=env_name,
platform=package_platform,
config=EnvironmentConfig(channels=tuple(env.config.channels)),
explicit_packages=list(env.explicit_packages),
external_packages=dict(env.external_packages),
)
validate_urls(cast("Environment", validation_env), FORMAT)
if env_name not in environments:
environments[env_name] = {
"channels": [
{"url": redact_channel_name(str(channel))}
for channel in env.config.channels
],
"packages": {},
}
platform_refs: list[dict[str, str]] = []
for pkg in sorted(env.explicit_packages, key=lambda p: p.name):
package_url = redact_url(pkg.url)
platform_refs.append({"conda": package_url})
package_kwargs: dict[str, Any] = {"conda": package_url}
for metadata_field in (
"build_number",
"sha256",
"md5",
"depends",
"constrains",
"features",
"track_features",
"license",
"license_family",
"size",
"python_site_packages_path",
):
value = pkg.get(metadata_field, None)
if metadata_field == "features" and isinstance(
value, (list, tuple)
):
value = " ".join(value)
if value is not None and (
value
or metadata_field
in {
"build_number",
"depends",
"constrains",
"features",
"track_features",
}
):
package_kwargs[metadata_field] = value
package_kwargs = cls.redact_data_urls({"packages": [package_kwargs]})[
"packages"
][0]
packages.append(
WorkspaceLockPackage(**package_kwargs).model_dump(exclude_none=True)
)
for manager, urls in env.external_packages.items():
for url in urls:
platform_refs.append({manager: redact_url(url)})
environments[env_name]["packages"][platform] = platform_refs
packages_by_url: dict[str, dict[str, Any]] = {}
for package in packages:
url = package["conda"]
packages_by_url[url] = cls.merge_package_records(
packages_by_url.get(url, {}),
package,
url,
)
return cls.redact_data_urls(
{
"version": LOCKFILE_VERSION,
"environments": environments,
"packages": list(packages_by_url.values()),
}
)
def render_lockfile(
ctx: WorkspaceContext,
resolved_envs: dict[str, ResolvedEnvironment],
*,
config: WorkspaceConfig | None = None,
platforms: tuple[str, ...] | None = None,
progress: Callable[[str, str], None] | None = None,
skip_unsolvable: bool = False,
on_skip: Callable[[str, str, SolveError], None] | None = None,
baseline_data: dict[str, Any] | None = None,
update_targets: Mapping[tuple[str, str], set[str]] | None = None,
preserve_existing: bool = False,
dry_run: bool = False,
) -> str:
"""Solve workspace environments and serialize ``conda.lock`` content.
Each environment in *resolved_envs* is solved in an empty temporary
prefix for every declared platform, intersected with *platforms* when
given. Selective updates seed the prefix from that target's locked
packages. When *config* is supplied, each ``(environment, platform)``
pair is resolved from the manifest just before solving, so each target
uses its own dependency tables. Serialisation is delegated to
:func:`.export.multiplatform_export` so this function and ``conda
export --format=conda-workspaces-lock-v1`` produce byte-identical
output. Solver chatter is silenced inside
:meth:`ResolvedEnvironment.solve_for_platform` itself, so the caller
is free to render status through the optional *progress* callback
without stdout bookkeeping.
Fails fast by default: the first unsolvable ``(environment,
platform)`` pair raises :class:`SolveError` with the platform
named, and no lockfile is written. When *skip_unsolvable* is
true, solver failures on an individual pair are reported via
*on_skip* (if given) and the lockfile continues with the remaining
pairs; :class:`AllTargetsUnsolvableError` is raised only if every
pair fails. Non-solver errors (missing channel, invalid manifest,
etc.) always abort.
Returns the serialized lockfile without writing it. When *dry_run* is
true, solver package-cache writes use disposable storage.
*preserve_existing* retains declared solutions from *baseline_data*,
solves missing targets, and removes undeclared targets. The caller must
validate the baseline against the previous manifest first. This mode
requires the complete environment set and cannot be combined with
selective updates, a platform filter, or skipped solver failures.
"""
from conda.common.serialize.yaml import dumps as yaml_dumps
from conda.models.environment import EnvironmentConfig
from .export import multiplatform_export
from .resolver import resolve_environment
host_platform = ctx.platform
envs: list[_SolvedEnvironment] = []
failures: list[SolveError] = []
if update_targets is not None and baseline_data is None:
raise ValueError("Selective lock updates require baseline lockfile data")
if preserve_existing:
if baseline_data is None:
raise ValueError("Lock preservation requires baseline lockfile data")
if update_targets is not None or platforms is not None or skip_unsolvable:
raise ValueError(
"Lock preservation requires complete platform reconciliation"
)
baseline = {} if baseline_data is None else baseline_data
baseline_loader: CondaLockLoader | None = None
if update_targets is not None or preserve_existing:
baseline_loader = CondaLockLoader(lockfile_path(ctx), data=baseline)
baseline = baseline_loader._data
platforms_by_environment = {
name: sorted(set(resolved.platforms or [host_platform]))
for name, resolved in resolved_envs.items()
}
solved_targets: set[tuple[str, str]] = set()
with isolated_package_cache(dry_run):
for name, resolved in resolved_envs.items():
declared = platforms_by_environment[name]
if update_targets is not None:
targets = [
target for target in declared if update_targets.get((name, target))
]
elif preserve_existing:
locked = (
baseline.get("environments", {}).get(name, {}).get("packages", {})
)
targets = [target for target in declared if target not in locked]
elif platforms is None:
targets = declared
else:
targets = []
for requested in platforms:
try:
target = resolved.resolve_platform_name(requested, declared)
except PlatformError:
continue
if target not in targets:
targets.append(target)
if not targets:
continue
for target in targets:
if progress is not None:
progress(name, target)
target_resolved = (
resolve_environment(config, name, target)
if config is not None
else resolved
)
package_platform = target_resolved.platform_subdir(target)
channels = tuple(
redact_channel_url(ch) for ch in target_resolved.channels
)
try:
with tempfile.TemporaryDirectory(
prefix="conda-workspaces-lock-"
) as temp_dir:
prefix = Path(temp_dir)
try:
if update_targets is not None:
solved_targets.add((name, target))
update_names = update_targets[(name, target)]
requested_specs = list(
target_resolved.conda_dependencies.values()
)
from .envs import _build_pypi_specs
requested_specs.extend(
_build_pypi_specs(target_resolved)
)
try:
records = CondaLockLoader.seed_prefix_from_data(
baseline_loader
if baseline_loader is not None
else baseline,
name,
target,
prefix,
requested_specs,
package_platform=package_platform,
)
except ValueError as exc:
raise LockfileIntegrityError(
lockfile_path(ctx),
str(exc),
) from exc
installed_names = {record.name for record in records}
missing = update_names - installed_names
if missing:
names = ", ".join(sorted(missing))
raise LockfileIntegrityError(
lockfile_path(ctx),
f"environment {name!r} on {target!r} is missing"
f" requested roots: {names}",
)
records = target_resolved.solve_for_platform(
package_platform,
prefix=prefix,
update_names=update_names,
)
else:
records = target_resolved.solve_for_platform(
package_platform,
prefix=prefix,
)
finally:
delete_prefix_from_linked_data(prefix)
except SolveError as exc:
if update_targets is not None or not skip_unsolvable:
raise
failures.append(exc)
if on_skip is not None:
on_skip(name, target, exc)
continue
envs.append(
_SolvedEnvironment(
name=name,
platform=target,
package_platform=package_platform,
config=EnvironmentConfig(channels=channels),
explicit_packages=records,
)
)
if update_targets is not None or preserve_existing:
if update_targets is not None:
missing_targets = {
target for target, names in update_targets.items() if names
} - solved_targets
if missing_targets:
targets = ", ".join(
f"{name}/{platform}" for name, platform in sorted(missing_targets)
)
raise ValueError(f"Selective lock targets are not declared: {targets}")
updated = CondaLockLoader.replace_solutions(
baseline,
envs,
platforms_by_environment=(
platforms_by_environment if preserve_existing else None
),
)
if config is not None:
updated_loader = CondaLockLoader(lockfile_path(ctx), data=updated)
status = updated_loader.check_targets(
config,
match_budget_scope="platform",
default_platform=host_platform,
)
if status.status != LockfileStatus.UP_TO_DATE:
raise LockfileStaleError(
config.manifest_path,
lockfile_path(ctx),
reason=status.reason,
)
return yaml_dumps(updated)
if failures and not envs:
raise AllTargetsUnsolvableError(failures)
return multiplatform_export(cast("Iterable[Environment]", envs))
def generate_lockfile(
ctx: WorkspaceContext,
resolved_envs: dict[str, ResolvedEnvironment],
*,
config: WorkspaceConfig | None = None,
platforms: tuple[str, ...] | None = None,
progress: Callable[[str, str], None] | None = None,
skip_unsolvable: bool = False,
on_skip: Callable[[str, str, SolveError], None] | None = None,
output_path: Path | None = None,
dry_run: bool = False,
publish_lockfile: Callable[[str], None] | None = None,
) -> Path:
"""Solve workspace environments and write their ``conda.lock``.
When *output_path* is given, the lockfile is written there instead
of the default ``<workspace>/conda.lock``. Matrix CI runners use this
to emit per-platform fragments that a coordinator job later combines
with :func:`merge_lockfiles`. When *dry_run* is true, solving and
serialization still run without creating or changing the output.
*publish_lockfile* can serialize canonical publication with another
workspace state file after every solve succeeds.
"""
path = output_path if output_path is not None else lockfile_path(ctx)
validate_lockfile_output(ctx, path)
output_generation = regular_file_generation(path)
content = render_lockfile(
ctx,
resolved_envs,
config=config,
platforms=platforms,
progress=progress,
skip_unsolvable=skip_unsolvable,
on_skip=on_skip,
dry_run=dry_run,
)
if dry_run:
return path
if publish_lockfile is not None:
publish_lockfile(content)
return path
return write_lockfile(
ctx,
content,
output_path=path,
expected_generation=output_generation,
)
def write_lockfile(
ctx: WorkspaceContext,
content: str,
*,
output_path: Path | None = None,
expected_generation: FileGeneration | None | object = (
_CURRENT_LOCKFILE_OUTPUT_GENERATION
),
) -> Path:
"""Validate and write already-rendered canonical lockfile content."""
path = output_path if output_path is not None else lockfile_path(ctx)
validate_lockfile_output(ctx, path)
if expected_generation is _CURRENT_LOCKFILE_OUTPUT_GENERATION:
expected_generation = regular_file_generation(path)
atomic_write_text(path, content, expected_generation=expected_generation)
return path
def merge_lockfiles(
paths: Sequence[Path],
ctx: WorkspaceContext,
*,
dry_run: bool = False,
publish_lockfile: Callable[[str], None] | None = None,
) -> Path:
"""Merge per-platform ``conda.lock`` fragments into a single lockfile.
Designed for CI matrix pipelines that split locking across runners:
each runner produces a fragment for one platform (typically via
``conda workspace lock --platform <subdir> --output
conda.lock.<subdir>``), and a coordinator job stitches them back
into a single ``<workspace>/conda.lock`` with this function.
Every fragment must be a ``version: 1`` lockfile. When the current
workspace manifest declares the fragment environment, its channel
list is canonical: fragments may omit unused manifest channels, but
any channels they do declare must be an ordered subset of the
manifest's list. Fragment environments not declared in the manifest
keep the stricter legacy rule and must agree on their ``channels``
list entry-for-entry and in the same order. Two fragments may not
both carry entries for the same ``(environment, platform)`` pair —
overlapping platforms indicate a misconfigured pipeline rather than
a legitimate merge. Any violation raises
:class:`LockfileMergeError` and nothing is written.
The merge happens at the YAML layer — going through
:class:`CondaLockLoader` would force
:func:`conda_lockfiles.records_from_conda_urls.records_from_conda_urls`
to fetch every package to populate :class:`PackageRecord` objects,
which defeats the purpose of merging cached fragments. We stitch
the dicts back together and hand them to conda's YAML dumper.
The output is byte-stable with a single-run
:func:`generate_lockfile` call over the same inputs: environments
are walked in first-seen order across fragments, platforms in
alphabetical order within each environment, and top-level
``packages`` are emitted in the same order
:meth:`CondaLockLoader.compose` would produce. Duplicate top-level
records for the same package URL are accepted only when their
metadata matches exactly.
When *dry_run* is true, every fragment is validated and the merged
data is serialized, but the canonical lockfile remains unchanged.
*publish_lockfile* can serialize the canonical write with another
workspace state file.
Returns the path to the merged lockfile.
"""
from conda.common.serialize.yaml import dump as yaml_dump
if not paths:
raise LockfileMergeError("no lockfile fragments were supplied")
out_path = lockfile_path(ctx)
validate_lockfile_output(ctx, out_path)
output_generation = regular_file_generation(out_path)
env_order: list[str] = []
env_channels: dict[str, list[dict[str, Any]]] = {}
env_platforms: dict[str, dict[str, list[dict[str, Any]]]] = {}
seen_pairs: dict[tuple[str, str], Path] = {}
packages_by_url: dict[str, dict[str, Any]] = {}
package_sources_by_url: dict[str, Path] = {}
manifest_env_channels: dict[str, list[dict[str, str]]] = {}
manifest_env_channel_urls: dict[str, list[str]] = {}
equal_package_record_pairs: dict[tuple[int, int], tuple[object, object]] = {}
equal_channel_pairs: dict[tuple[int, int], tuple[object, object]] = {}
fragment_bytes = 0
fragment_items = 0
fragment_strings = DocumentStringBudget(MAX_LOCKFILE_BYTES)
for env_name, env_obj in ctx.config.environments.items():
entries = _manifest_channel_entries(ctx.config, env_obj)
manifest_env_channels[env_name] = entries
manifest_env_channel_urls[env_name] = _normalize_channel_urls(
entry["url"] for entry in entries
)
for path in paths:
if path.is_symlink() or not path.is_file():
raise LockfileMergeError(f"fragment '{path}' does not exist")
try:
content = read_regular_file_bytes(
path,
maximum_bytes=MAX_LOCKFILE_BYTES,
label="lockfile fragment",
)
except ValueError as exc:
raise LockfileMergeError(
f"fragment '{path}' cannot be read safely"
) from exc
fragment_bytes += len(content)
if fragment_bytes > MAX_LOCKFILE_BYTES:
raise LockfileMergeError(
"lockfile fragments exceed the aggregate maximum size of "
f"{MAX_LOCKFILE_BYTES:,} bytes"
)
try:
data = load_lockfile_data(content, string_budget=fragment_strings)
fragment_items += validate_document_limits(
data,
label="Lockfile fragments",
maximum_depth=MAX_LOCKFILE_DEPTH,
maximum_collection_items=MAX_LOCKFILE_COLLECTION_ITEMS,
maximum_items=MAX_LOCKFILE_ITEMS,
)
if fragment_items > MAX_LOCKFILE_ITEMS:
raise ValueError(
"Lockfile fragments contain more than "
f"{MAX_LOCKFILE_ITEMS:,} aggregate collection items"
)
data = CondaLockLoader.redact_data_urls(data)
version = data.get("version")
if type(version) is not int:
raise ValueError("Lockfile version must be an integer")
except ValueError as exc:
raise LockfileMergeError(str(exc)) from exc
if version != LOCKFILE_VERSION:
raise LockfileMergeError(
f"fragment '{path}' has version {version!r}, "
f"expected {LOCKFILE_VERSION}"
)
for record in data.get("packages", []) or []:
url = record.get("url") or record.get("conda") or record.get("pypi")
if not url:
continue
existing = packages_by_url.get(url)
if existing is None:
packages_by_url[url] = record
package_sources_by_url[url] = path
elif not document_values_equal(
existing,
record,
equal_value_pairs=equal_package_record_pairs,
):
raise LockfileMergeError(
f"fragment '{path}' has a conflicting package record for "
f"URL '{redact_url(url)}'",
hints=[
(
"The same package URL must have identical top-level"
" metadata in every fragment."
),
f"The first record came from '{package_sources_by_url[url]}'.",
],
)
for env_name, env_data in (data.get("environments") or {}).items():
channels = env_data.get("channels")
if channels is None:
channels = []
if not isinstance(channels, list) or not all(
isinstance(channel, dict)
and isinstance(channel.get("url"), str)
and channel["url"]
for channel in channels
):
raise LockfileMergeError(
f"Lockfile environment {env_name!r} has invalid channels"
)
manifest_channels = manifest_env_channels.get(env_name)
if manifest_channels is not None:
manifest_urls = manifest_env_channel_urls[env_name]
fragment_urls = _normalize_channel_urls(
entry["url"] for entry in channels
)
next_index = 0
for url in fragment_urls:
try:
next_index = manifest_urls.index(url, next_index) + 1
except ValueError as exc:
raise LockfileMergeError(
f"environment '{env_name}' channels differ between "
f"fragment '{path}' and the manifest",
hints=[
(
"Every fragment channel list must be an ordered"
" subset of the manifest channel list for a"
" shared environment."
),
],
) from exc
channels = list(manifest_channels)
existing = env_channels.get(env_name)
if existing is None:
env_order.append(env_name)
env_channels[env_name] = channels
env_platforms[env_name] = {}
elif not document_values_equal(
existing,
channels,
equal_value_pairs=equal_channel_pairs,
):
raise LockfileMergeError(
f"environment '{env_name}' channels differ between "
f"fragments; '{path}' disagrees with an earlier fragment",
hints=[
(
"Every fragment must declare the same channel list"
" (same entries, same order) for a shared environment."
),
],
)
for platform, refs in (env_data.get("packages") or {}).items():
pair = (env_name, platform)
if pair in seen_pairs:
raise LockfileMergeError(
f"environment '{env_name}' on platform "
f"'{platform}' is present in both "
f"'{seen_pairs[pair]}' and '{path}'",
hints=[
(
"Each (environment, platform) pair must come"
" from exactly one fragment."
),
],
)
seen_pairs[pair] = path
package_platform = ctx.config.platform_subdir(platform)
try:
channel_urls = CondaLockLoader.channel_urls_for_env_data(
{"channels": channels},
package_platform,
)
channel_locations = CondaLockLoader._channel_locations(channel_urls)
except ValueError as exc:
raise LockfileMergeError(
f"environment '{env_name}' channels in fragment "
f"'{path}' cannot be resolved"
) from exc
validated_refs: list[dict[str, Any]] = []
for ref in refs or []:
if not isinstance(ref, dict) or set(ref) != {"conda"}:
raise LockfileMergeError(
f"environment '{env_name}' package refs on "
f"'{platform}' in fragment '{path}' must contain "
"exactly one 'conda' entry"
)
url = ref["conda"]
if not isinstance(url, str):
raise LockfileMergeError(
f"environment '{env_name}' contains an invalid "
f"package ref on '{platform}' in fragment '{path}'"
)
if not CondaLockLoader._url_matches_locations(
url, channel_locations
):
raise LockfileMergeError(
f"package URL '{redact_url(url)}' in environment "
f"'{env_name}' on '{platform}' is not under any "
"declared channel"
)
record = packages_by_url.get(url)
if record is None:
raise LockfileMergeError(
f"package URL '{redact_url(url)}' in environment "
f"'{env_name}' on '{platform}' has no top-level "
"package record"
)
try:
CondaLockLoader.digest_fragment_for_record(record, url)
except ValueError as exc:
raise LockfileMergeError(str(exc)) from exc
validated_refs.append(ref)
env_platforms[env_name][platform] = validated_refs
# Rebuild top-level ``packages`` in the same order
# :meth:`CondaLockLoader.compose` would produce for a single-run
# solve: iterate envs in first-seen order, platforms alphabetically,
# then each platform's refs (already sorted by package name by the
# producing fragment).
merged_packages: list[dict[str, Any]] = []
emitted_urls: set[str] = set()
for env_name in env_order:
for platform in sorted(env_platforms[env_name]):
for ref in env_platforms[env_name][platform]:
url = ref.get("conda")
if not url or url in emitted_urls:
continue
record = packages_by_url.get(url)
if record is not None:
merged_packages.append(record)
emitted_urls.add(url)
merged = {
"version": LOCKFILE_VERSION,
"environments": {
name: {
"channels": env_channels[name],
"packages": {
platform: env_platforms[name][platform]
for platform in sorted(env_platforms[name])
},
}
for name in env_order
},
"packages": merged_packages,
}
try:
validate_document_limits(
merged,
label="Merged lockfile",
maximum_depth=MAX_LOCKFILE_DEPTH,
maximum_collection_items=MAX_LOCKFILE_COLLECTION_ITEMS,
maximum_items=MAX_LOCKFILE_ITEMS,
)
merged = CondaLockLoader.redact_data_urls(merged)
except ValueError as exc:
raise LockfileMergeError(str(exc)) from exc
buf = io.StringIO()
yaml_dump(merged, buf)
try:
content = decode_limited_text(
buf.getvalue(),
maximum_bytes=MAX_LOCKFILE_BYTES,
label="Merged lockfile",
)
except ValueError as exc:
raise LockfileMergeError(str(exc)) from exc
if dry_run:
return out_path
if publish_lockfile is not None:
publish_lockfile(content)
else:
atomic_write_text(
out_path,
content,
expected_generation=output_generation,
)
return out_path
def validate_lockfile_output(ctx: WorkspaceContext, path: Path) -> None:
"""Reject lock outputs that could overwrite the selected manifest."""
manifest_path = Path(ctx.config.manifest_path)
if output_paths_collide(path, manifest_path):
raise ValueError("Lockfile output cannot overwrite the workspace manifest.")
if path.is_symlink():
raise ValueError("Lockfile output cannot be a symbolic link.")
validate_file_output(path)
@dataclass
class LockfileInstallPlan:
"""A fully validated exact install that can be executed without refetching."""
env_name: str
prefix: Path
target_prefix_override: str | Path | None
records: list[PackageRecord]
requested_specs: list[str] | None
prune_setup: PrefixSetup | None
resolved: ResolvedEnvironment | None
update_path_dependencies: bool
preflight_prefix_identity: tuple[int, int] | None
expected_prefix_identity: tuple[int, int] | None
validate_workspace: Callable[[], None] | None
def validate_virtual_packages(self, platform: str) -> None:
"""Check required and optional virtual constraints against this machine."""
from conda.base.context import context as conda_context
from conda.models.match_spec import MatchSpec
requirements = [
(MatchSpec(spec), required)
for record in self.records
for specs, required in ((record.depends, True), (record.constrains, False))
for spec in specs
if spec.startswith("__")
]
if self.resolved is not None:
requirements.extend(
(spec, True)
for spec in (
*self.resolved.conda_dependencies.values(),
*self.resolved.system_requirement_specs(),
)
if spec.name and spec.name.startswith("__")
)
if not requirements:
return
with conda_context._override("_subdir", platform):
actual = {
record.name: record
for record in conda_context.plugin_manager.get_virtual_package_records()
}
for requirement, required in requirements:
record = actual.get(requirement.name)
if (record is None and required) or (
record is not None and not requirement.match(record)
):
raise CondaWorkspacesError(
f"This machine does not satisfy {requirement} "
f"required by environment '{self.env_name}'."
)
def validate_workspace_generation(self) -> None:
"""Revalidate a guarded workspace immediately before path consumers."""
if self.validate_workspace is not None:
self.validate_workspace()
def remove_preflight_prefix(self, ctx: WorkspaceContext) -> None:
"""Remove the managed prefix generation inspected during preflight."""
identity = self.preflight_prefix_identity
if identity is None:
return
self.validate_workspace_generation()
managed_prefix = canonicalize_system_path_alias(ctx.env_prefix(self.env_name))
if self.prefix != managed_prefix:
raise CondaWorkspacesError(
"Cannot replace an explicit prefix through a workspace install."
)
from .envs import remove_environment
remove_environment(
ctx,
self.env_name,
expected_prefix_identity=identity,
)
@staticmethod
def prefix_identity(path: Path) -> tuple[int, int] | None:
"""Return one regular directory generation without following links."""
try:
current = path.lstat()
except FileNotFoundError:
return None
if not stat.S_ISDIR(current.st_mode):
raise CondaWorkspacesError(
f"Environment prefix is not a regular directory: {path}"
)
return current.st_dev, current.st_ino
@classmethod
def prepare(
cls,
ctx: WorkspaceContext,
env_name: str,
*,
prefix: Path | None = None,
platform: str | None = None,
target_prefix_override: str | Path | None = None,
lockfile_data: dict[str, Any] | None = None,
lockfile_loader: CondaLockLoader | None = None,
update_names: set[str] | None = None,
prune: bool = True,
replace_existing: bool = False,
require_absent: bool = False,
validate_workspace: Callable[[], None] | None = None,
) -> LockfileInstallPlan:
"""Fetch packages and validate every non-mutating install input."""
from conda.base.context import context as conda_context
from conda.core.link import PrefixSetup
from conda.core.prefix_data import PrefixData
from conda.history import History
from conda.misc import get_package_records_from_explicit
from conda.models.match_spec import MatchSpec
path = lockfile_path(ctx)
if lockfile_loader is not None and lockfile_data is not None:
raise ValueError("Provide either lockfile data or a prepared loader")
if lockfile_loader is None:
if lockfile_data is None:
try:
if validate_workspace is not None:
validate_workspace()
lockfile_data = load_lockfile_data(
read_regular_file_bytes(
path,
maximum_bytes=MAX_LOCKFILE_BYTES,
label="workspace lockfile",
)
)
if validate_workspace is not None:
validate_workspace()
except (OSError, ValueError) as exc:
raise LockfileNotFoundError("(all)", path) from exc
lockfile_loader = CondaLockLoader(path, data=lockfile_data)
lock_platform = platform or ctx.platform
package_platform = ctx.config.platform_subdir(lock_platform)
resolved: ResolvedEnvironment | None = None
try:
from .resolver import resolve_environment
try:
resolved = resolve_environment(
ctx.config, env_name, lock_platform
).with_absolute_paths(Path(ctx.config.root))
lock_platform = ctx.config.resolve_platform_name(
lock_platform,
resolved.platforms or ctx.config.platforms,
)
package_platform = ctx.config.platform_subdir(lock_platform)
except (EnvironmentNotFoundError, PlatformError):
if platform is not None:
raise
pass
if (
platform is not None
and package_platform != conda_context._native_subdir()
):
raise CondaWorkspacesError(
f"Cannot install platform '{lock_platform}' on this machine "
f"({conda_context._native_subdir()})."
)
urls = lockfile_loader.explicit_package_specs_for(
lock_platform,
env_name,
package_platform=package_platform,
)
except (ValueError, OSError) as exc:
raise LockfileNotFoundError(env_name, path) from exc
install_prefix = canonicalize_system_path_alias(
prefix or ctx.env_prefix(env_name)
)
initial_prefix_identity = cls.prefix_identity(install_prefix)
if require_absent and initial_prefix_identity is not None:
raise CondaWorkspacesError(
f"Workspace environment prefix already exists: {install_prefix}"
)
validate_directory_output(install_prefix)
override = (
conda_context._override(
"target_prefix_override",
str(target_prefix_override),
)
if target_prefix_override is not None
else nullcontext()
)
# Preparation only fetches packages. CLI dry runs isolate those cache
# writes and must still finish validation before skipping installation.
with override, conda_context._override("dry_run", False):
if validate_workspace is not None:
validate_workspace()
records = cast(
"list[PackageRecord]",
list(get_package_records_from_explicit(urls)),
)
if validate_workspace is not None:
validate_workspace()
requested_specs: list[MatchSpec] | None = None
unlocked_requested_names: set[str | None] = set()
if resolved is not None:
from .envs import (
_build_pypi_specs,
validate_path_dependencies,
)
requested_specs = list(resolved.conda_dependencies.values())
requested_specs.extend(_build_pypi_specs(resolved))
for dependency in resolved.pypi_dependencies.values():
if dependency.path:
spec = MatchSpec(dependency.name)
requested_specs.append(spec)
unlocked_requested_names.add(spec.name)
if update_names is None:
validate_path_dependencies(resolved)
prune_setup = None
existing_prefix = (
anchored_directory(install_prefix)
if initial_prefix_identity is not None
else nullcontext(None)
)
if validate_workspace is not None:
validate_workspace()
with existing_prefix:
if resolved is not None:
from .envs import validate_activation_metadata
validate_activation_metadata(install_prefix, resolved)
if initial_prefix_identity is not None and (
cls.prefix_identity(install_prefix) != initial_prefix_identity
):
raise CondaWorkspacesError(
f"Environment prefix changed during preflight: {install_prefix}"
)
prefix_data = PrefixData(str(install_prefix))
if prefix_data.is_environment():
locked_names = {record.name for record in records}
requested_names = (
{spec.name for spec in requested_specs}
if requested_specs is not None
else None
)
stale_requests = (
tuple(
spec
for name, spec in History(str(install_prefix))
.get_requested_specs_map()
.items()
if name not in requested_names
)
if requested_names is not None
else ()
)
extras = tuple(
record
for record in prefix_data.iter_records()
if record.name not in locked_names
and record.name not in unlocked_requested_names
)
if prune and not replace_existing and (extras or stale_requests):
prune_setup = PrefixSetup(
str(install_prefix),
extras,
(),
stale_requests,
(),
(),
)
if initial_prefix_identity is not None and (
cls.prefix_identity(install_prefix) != initial_prefix_identity
):
raise CondaWorkspacesError(
f"Environment prefix changed during preflight: {install_prefix}"
)
if validate_workspace is not None:
validate_workspace()
if require_absent and cls.prefix_identity(install_prefix) is not None:
raise CondaWorkspacesError(
f"Workspace environment prefix already exists: {install_prefix}"
)
plan = cls(
env_name=env_name,
prefix=install_prefix,
target_prefix_override=target_prefix_override,
records=records,
requested_specs=(
[str(spec) for spec in requested_specs]
if requested_specs is not None
else None
),
prune_setup=prune_setup,
resolved=resolved,
update_path_dependencies=update_names is None,
preflight_prefix_identity=initial_prefix_identity,
expected_prefix_identity=(
None if replace_existing else initial_prefix_identity
),
validate_workspace=validate_workspace,
)
plan.validate_virtual_packages(package_platform)
return plan
@contextmanager
def open_prefix(self) -> Iterator[Callable[[], None]]:
"""Create or reopen the planned prefix through an anchored parent."""
self.validate_workspace_generation()
expected = self.expected_prefix_identity
with anchored_directory(self.prefix.parent, create=True) as parent_descriptor:
if parent_descriptor is None:
current = self.prefix_identity(self.prefix)
if current is None:
if expected is not None:
raise CondaWorkspacesError(
f"Environment prefix changed before install: {self.prefix}"
)
self.prefix.mkdir()
current = self.prefix_identity(self.prefix)
elif current != expected:
raise CondaWorkspacesError(
f"Environment prefix changed before install: {self.prefix}"
)
assert current is not None
def require_current_prefix() -> None:
self.validate_workspace_generation()
if self.prefix_identity(self.prefix) != current:
raise CondaWorkspacesError(
f"Environment prefix changed during install: {self.prefix}"
)
yield require_current_prefix
return
opened_parent = os.fstat(parent_descriptor)
try:
current = os.stat(
self.prefix.name,
dir_fd=parent_descriptor,
follow_symlinks=False,
)
except FileNotFoundError:
if expected is not None:
raise CondaWorkspacesError(
f"Environment prefix changed before install: {self.prefix}"
) from None
os.mkdir(self.prefix.name, dir_fd=parent_descriptor)
current = os.stat(
self.prefix.name,
dir_fd=parent_descriptor,
follow_symlinks=False,
)
else:
identity = current.st_dev, current.st_ino
if expected is None or identity != expected:
raise CondaWorkspacesError(
f"Environment prefix changed before install: {self.prefix}"
)
if not stat.S_ISDIR(current.st_mode):
raise CondaWorkspacesError(
f"Environment prefix is not a regular directory: {self.prefix}"
)
prefix_flags = (
os.O_RDONLY
| getattr(os, "O_DIRECTORY", 0)
| getattr(os, "O_CLOEXEC", 0)
| getattr(os, "O_NOFOLLOW", 0)
)
prefix_descriptor = os.open(
self.prefix.name,
prefix_flags,
dir_fd=parent_descriptor,
)
try:
opened_prefix = os.fstat(prefix_descriptor)
identity = opened_prefix.st_dev, opened_prefix.st_ino
if identity != (current.st_dev, current.st_ino):
raise CondaWorkspacesError(
f"Environment prefix changed before install: {self.prefix}"
)
def require_current_prefix() -> None:
self.validate_workspace_generation()
try:
parent_entry = os.stat(
self.prefix.name,
dir_fd=parent_descriptor,
follow_symlinks=False,
)
live_parent = self.prefix.parent.lstat()
live_prefix = self.prefix.lstat()
except OSError as exc:
raise CondaWorkspacesError(
f"Environment prefix changed during install: {self.prefix}"
) from exc
if (
not stat.S_ISDIR(parent_entry.st_mode)
or (parent_entry.st_dev, parent_entry.st_ino) != identity
or not stat.S_ISDIR(live_parent.st_mode)
or (live_parent.st_dev, live_parent.st_ino)
!= (opened_parent.st_dev, opened_parent.st_ino)
or not stat.S_ISDIR(live_prefix.st_mode)
or (live_prefix.st_dev, live_prefix.st_ino) != identity
):
raise CondaWorkspacesError(
f"Environment prefix changed during install: {self.prefix}"
)
require_current_prefix()
yield require_current_prefix
finally:
os.close(prefix_descriptor)
def execute(self) -> None:
"""Execute this plan while retaining its fetched package records."""
from conda.base.context import context as conda_context
from conda.core.link import UnlinkLinkTransaction
from conda.misc import install_explicit_packages
override = (
conda_context._override(
"target_prefix_override",
str(self.target_prefix_override),
)
if self.target_prefix_override is not None
else nullcontext()
)
with override, self.open_prefix() as require_current_prefix:
require_current_prefix()
if self.prune_setup is not None:
UnlinkLinkTransaction(self.prune_setup).execute()
require_current_prefix()
install_explicit_packages(
package_cache_records=self.records,
prefix=str(self.prefix),
requested_specs=self.requested_specs,
)
require_current_prefix()
if self.resolved is not None:
from .envs import (
_apply_activation_env,
_apply_activation_scripts,
_install_path_deps,
)
_apply_activation_env(self.prefix, self.resolved.activation_env)
require_current_prefix()
_apply_activation_scripts(
self.prefix,
self.resolved.activation_scripts,
)
require_current_prefix()
if self.update_path_dependencies:
_install_path_deps(self.prefix, self.resolved)
require_current_prefix()
sys.stdout.flush()
def install_from_lockfile(
ctx: WorkspaceContext,
env_name: str,
*,
prefix: Path | None = None,
platform: str | None = None,
target_prefix_override: str | Path | None = None,
dry_run: bool = False,
lockfile_data: dict[str, Any] | None = None,
update_names: set[str] | None = None,
prune: bool = True,
replace_existing: bool = False,
validate_workspace: Callable[[], None] | None = None,
) -> LockfileInstallPlan:
"""Install an environment from ``conda.lock``.
Reads the lockfile via :class:`CondaLockLoader`, extracts the
package list for *env_name* on the current platform, downloads the
exact packages, optionally removes conda packages absent from the lock,
and installs them into the environment prefix without a solver.
Requested-spec history is reconciled to the resolved manifest
roots.
When *dry_run* is true, package records are fetched and the requested
specs, prune transaction, activation inputs, and local project inputs are
prepared without creating or changing the target prefix.
*lockfile_data* installs a previously rendered in-memory solution so
solving and installation cannot diverge. *update_names* suppresses
rebuilding unrelated local path dependencies during selective updates.
When *prune* is false, packages and requested specs absent from the lock
are preserved. *replace_existing* removes the exact managed prefix
generation inspected during preparation before recreating it. A dry run
only prepares that replacement. It cannot be combined with *prefix*.
Raises ``LockfileNotFoundError`` if the lockfile is missing or does
not contain the requested environment/platform.
"""
if replace_existing and prefix is not None:
raise CondaWorkspacesError(
"Prefix replacement cannot be combined with an explicit prefix."
)
plan = LockfileInstallPlan.prepare(
ctx,
env_name,
prefix=prefix,
platform=platform,
target_prefix_override=target_prefix_override,
lockfile_data=lockfile_data,
update_names=update_names,
prune=prune,
replace_existing=replace_existing,
validate_workspace=validate_workspace,
)
if not dry_run:
if replace_existing:
plan.remove_preflight_prefix(ctx)
plan.execute()
return plan
