"""Data models for workspace configuration.
These dataclasses represent the parsed workspace manifest in a
format-agnostic way. Parsers convert from pixi.toml / pyproject.toml /
conda.toml into these models; downstream code only works with
these types.
Conda dependencies use :class:`~conda.models.match_spec.MatchSpec`
directly, and channels use :class:`~conda.models.channel.Channel`,
so the workspace layer benefits from conda's own validation, URL
resolution, and spec parsing.
"""
from __future__ import annotations
import re
from collections.abc import Collection, Mapping
from dataclasses import dataclass, field, fields, replace
from pathlib import PurePosixPath
from typing import TYPE_CHECKING, ClassVar
from urllib.parse import unquote_to_bytes, urlsplit
if TYPE_CHECKING:
from collections.abc import Iterable
from typing import Any
from conda.base.constants import KNOWN_SUBDIRS
from conda.models.channel import Channel
from conda.models.match_spec import MatchSpec
from .exceptions import (
EnvironmentNameInvalidError,
EnvironmentNotFoundError,
FeatureNotFoundError,
PlatformError,
)
from .paths import is_path_segment, portable_path_key
_URL_CANDIDATE_RE = re.compile(r"(?i)(?:[a-z][a-z0-9+.-]*:)?//[^\s'\"<>]*")
def _decode_url_bytes(value: str) -> bytes | None:
decoded = value.encode()
for _ in range(4):
next_value = unquote_to_bytes(decoded)
if next_value == decoded:
return decoded
decoded = next_value
return None
def _redact_url_once(url: str) -> str:
"""Redact sensitive URL structure visible without decoding the whole value."""
absolute_like = re.match(r"(?i)^[a-z][a-z0-9+.-]*://", url) is not None
scheme_relative_like = url.startswith("//")
try:
parts = urlsplit(url)
except ValueError:
return "<redacted-url>" if absolute_like or scheme_relative_like else url
if absolute_like and not parts.netloc and parts.scheme != "file":
return "<redacted-url>"
if scheme_relative_like and not parts.netloc:
return "<redacted-url>"
if not parts.scheme and not scheme_relative_like:
return url
if parts.scheme != "file":
try:
hostname = parts.hostname
_ = parts.port
except ValueError:
return "<redacted-url>"
if (
not hostname
or "\\" in parts.netloc
or any(character.isspace() for character in parts.netloc)
):
return "<redacted-url>"
decoded_path = _decode_url_bytes(parts.path)
if decoded_path is None:
return "<redacted-url>"
if re.search(rb"(?i)(?:[a-z][a-z0-9+.-]*:)?//[^/]*@", decoded_path):
return "<redacted-url>"
segments: list[str] = []
redact_next = False
path_changed = False
for raw_segment in parts.path.split("/"):
decoded_segment = _decode_url_bytes(raw_segment)
if decoded_segment is None:
return "<redacted-url>"
normalized_segment = decoded_segment.lower()
if redact_next:
redact_next = False
path_changed = True
continue
if normalized_segment == b"t" or normalized_segment.endswith(b"/t"):
redact_next = True
path_changed = True
continue
if normalized_segment.startswith(b"t/") or b"/t/" in normalized_segment:
path_changed = True
continue
segments.append(raw_segment)
path = "/".join(segments) if path_changed else parts.path
redacted = parts._replace(
netloc=parts.netloc.rpartition("@")[2],
path=path,
query="",
fragment="",
).geturl()
separator = url.find(":")
if separator >= 0 and url[:separator].lower() == parts.scheme:
redacted = url[:separator] + redacted[separator:]
return redacted
[docs]
def redact_url(url: str) -> str:
"""Remove credentials, tokens, query, and fragment from an absolute URL."""
redacted = _redact_url_once(url)
decoded = _decode_url_bytes(redacted)
if decoded is None:
return "<redacted-url>"
try:
decoded_text = decoded.decode()
except UnicodeDecodeError:
return "<redacted-url>"
if decoded_text != redacted and _redact_url_once(decoded_text) != decoded_text:
return "<redacted-url>"
return redacted
[docs]
def has_url_credentials(value: str) -> bool:
"""Return whether a manifest string contains sensitive URL material."""
decoded = _decode_url_bytes(value)
if decoded is None:
return True
if re.search(
rb"(?i)(?:^|[^a-z0-9._~-])t[\\/][^\\/\s?#]+(?:[\\/]|$)",
decoded,
):
return True
candidates = [value]
if decoded != value.encode():
candidates.append(decoded.decode("utf-8", errors="ignore"))
return any(
redact_url(match.group(0)) != match.group(0)
for candidate in candidates
for match in _URL_CANDIDATE_RE.finditer(candidate)
)
[docs]
def has_match_spec_url_credentials(
spec: MatchSpec,
*,
include_channel: bool = True,
) -> bool:
"""Return whether a conda MatchSpec field contains sensitive URL material."""
return any(
value is not None and has_url_credentials(str(value))
for field in MatchSpec.FIELD_NAMES
if include_channel or field != "channel"
for value in (spec.get_raw_value(field),)
)
[docs]
def redact_url_text(value: str) -> str:
"""Remove sensitive URL material embedded in a larger diagnostic string."""
pieces: list[str] = []
offset = 0
for match in _URL_CANDIDATE_RE.finditer(value):
pieces.append(value[offset : match.start()])
pieces.append(redact_url(match.group(0)))
offset = match.end()
pieces.append(value[offset:])
redacted = "".join(pieces)
if has_url_credentials(redacted):
return "<redacted-url-value>"
return redacted
[docs]
def has_url_credentials_in_data(value: object) -> bool:
"""Return whether nested structured data contains URL credentials."""
values = [value]
seen: set[int] = set()
while values:
current = values.pop()
if isinstance(current, str):
if has_url_credentials(current):
return True
continue
if isinstance(current, Mapping):
identity = id(current)
if identity in seen:
continue
seen.add(identity)
values.extend(current.keys())
values.extend(current.values())
elif isinstance(current, Collection) and not isinstance(
current,
(bytes, bytearray),
):
identity = id(current)
if identity in seen:
continue
seen.add(identity)
values.extend(current)
return False
[docs]
def normalize_url_scheme(url: str) -> str:
"""Make URL schemes explicit and lowercase without changing URL identity."""
if url.startswith("//"):
try:
if urlsplit(url).netloc:
return f"https:{url}"
except ValueError:
return url
match = re.match(r"(?i)^([a-z][a-z0-9+.-]*)(?=://)", url)
if match is None:
return url
return match.group(1).lower() + url[match.end(1) :]
[docs]
def redact_channel_name(channel: str | Channel) -> str:
"""Return a credential-free channel name for display or export.
Conda treats a relative ``t/<token>/<channel>`` value as a named channel,
so its canonical name still contains the token. Resolve that form only
when redaction changes the resolved URL, while preserving ordinary names
such as ``conda-forge``.
"""
original = channel.canonical_name if isinstance(channel, Channel) else channel
if original.startswith("//"):
original = normalize_url_scheme(original)
redacted = redact_url(original)
if redacted != original:
return redact_url_text(redacted)
parsed = (
channel
if isinstance(channel, Channel)
else Channel(normalize_url_scheme(channel))
)
location = parsed.location or ""
if location.startswith("//"):
token_path = f"/t/{parsed.token}" if parsed.token else ""
return redact_url_text(
normalize_url_scheme(redact_url(f"{location}{token_path}/{parsed.name}"))
)
resolved = str(parsed)
redacted_resolved = redact_url(resolved)
candidate = redacted_resolved if redacted_resolved != resolved else original
return redact_url_text(candidate)
[docs]
def redact_channel_url(channel: Channel) -> str:
"""Return a credential-free resolved URL for a channel or channel name."""
if not isinstance(channel, Channel):
return redact_channel_name(str(channel))
canonical = redact_channel_name(channel)
if canonical != channel.canonical_name:
return canonical
value = (
canonical if re.match(r"(?i)^[a-z][a-z0-9+.-]*://", canonical) else str(channel)
)
return redact_url(value)
[docs]
@dataclass(frozen=True)
class LockfileStatus:
"""Status of the lockfile relative to the workspace manifest."""
UP_TO_DATE: ClassVar[str] = "up-to-date"
OUT_OF_DATE: ClassVar[str] = "out-of-date"
MISSING: ClassVar[str] = "missing"
status: str
reason: str = ""
def __post_init__(self) -> None:
object.__setattr__(self, "reason", redact_url_text(self.reason or ""))
[docs]
@dataclass(frozen=True)
class PyPIDependency:
"""A PyPI dependency (PEP 508 string).
Version-only dependencies are translated to conda equivalents via
``conda-pypi``'s grayskull mapping and merged into the same solver
call as conda deps. Local ``path`` dependencies are built and
installed post-solve via conda-pypi's build system. Git and URL
dependencies are parsed for pixi manifest compatibility but are not
installed yet.
"""
name: str
spec: str = ""
extras: tuple[str, ...] = ()
path: str | None = None
editable: bool = False
git: str | None = None
branch: str | None = None
tag: str | None = None
rev: str | None = None
url: str | None = None
def __str__(self) -> str:
base = self.name
if self.extras:
base = f"{base}[{','.join(self.extras)}]"
if self.git:
return f"{base} @ git+{self.git}"
if self.path:
prefix = "-e " if self.editable else ""
return f"{prefix}{base} @ {self.path}"
if self.url:
return f"{base} @ {self.url}"
if self.spec:
return f"{base}{self.spec}"
return base
[docs]
def to_toml(self) -> str | dict[str, object]:
"""Return this dependency in its manifest representation."""
fields: dict[str, object] = {
"version": self.spec,
"extras": list(self.extras),
"path": self.path,
"editable": self.editable,
"git": self.git,
"branch": self.branch,
"tag": self.tag,
"rev": self.rev,
"url": self.url,
}
return {key: value for key, value in fields.items() if value} or "*"
[docs]
def redacted(self) -> PyPIDependency:
"""Return a copy safe to include in diagnostic output."""
return replace(
self,
spec=redact_url_text(self.spec),
path=redact_url_text(self.path) if self.path else None,
git=redact_url_text(self.git) if self.git else None,
url=redact_url_text(self.url) if self.url else None,
)
[docs]
def to_manifest_toml(self) -> str | dict[str, object]:
"""Return a safe manifest value without changing direct URL semantics."""
if self.redacted() != self:
raise ValueError(
f"PyPI dependency '{self.name}' has a direct URL that cannot be"
" written safely. Remove embedded authentication, Anaconda token"
" paths, queries, and fragments, then configure authentication"
" outside the workspace manifest."
)
return self.to_toml()
[docs]
@dataclass
class Feature:
"""A composable group of dependencies and settings.
Features map directly to ``[feature.<name>]`` tables in a pixi manifest.
They can provide conda dependencies, PyPI dependencies, channel
overrides, platform restrictions, and environment variables.
The special feature named ``"default"`` corresponds to the top-level
workspace dependencies.
"""
DEFAULT_NAME: ClassVar[str] = "default"
name: str
conda_dependencies: dict[str, MatchSpec] = field(default_factory=dict)
pypi_dependencies: dict[str, PyPIDependency] = field(default_factory=dict)
channels: list[Channel] = field(default_factory=list)
platforms: list[str] = field(default_factory=list)
system_requirements: dict[str, str] = field(default_factory=dict)
activation_scripts: list[str] = field(default_factory=list)
activation_env: dict[str, str] = field(default_factory=dict)
# Per-platform overrides: platform -> deps
target_conda_dependencies: dict[str, dict[str, MatchSpec]] = field(
default_factory=dict
)
target_pypi_dependencies: dict[str, dict[str, PyPIDependency]] = field(
default_factory=dict
)
@property
def is_default(self) -> bool:
return self.name == self.DEFAULT_NAME
[docs]
@dataclass
class Environment:
"""A named environment composed from features and private dependencies.
This maps to a ``[environments]`` entry in a pixi manifest.
An environment inherits the ``default`` feature plus any additional
features listed in *features*. Dependencies declared directly on the
environment are private to it and override dependencies from its features.
*no_default_feature* can be set to exclude the default feature,
matching pixi's ``no-default-feature = true`` option.
"""
DEFAULT_NAME: ClassVar[str] = "default"
name: str
features: list[str] = field(default_factory=list)
no_default_feature: bool = False
conda_dependencies: dict[str, MatchSpec] = field(default_factory=dict)
pypi_dependencies: dict[str, PyPIDependency] = field(default_factory=dict)
target_conda_dependencies: dict[str, dict[str, MatchSpec]] = field(
default_factory=dict
)
target_pypi_dependencies: dict[str, dict[str, PyPIDependency]] = field(
default_factory=dict
)
@property
def is_default(self) -> bool:
return self.name == self.DEFAULT_NAME
[docs]
@dataclass(frozen=True)
class ArchiveConfig:
"""Archive settings from ``[workspace.archive]``."""
include: tuple[str, ...] = ()
exclude: tuple[str, ...] = ()
compression: str = "zst"
compression_level: int | None = None
[docs]
@dataclass
class WorkspaceConfig:
"""Complete parsed workspace configuration.
This is the top-level model that parsers produce. It contains
all channels, platforms, features, and environments defined in
a workspace manifest.
*manifest_path* points to the file that was parsed (for error
messages and relative path resolution).
"""
name: str | None = None
version: str | None = None
description: str | None = None
channels: list[Channel] = field(default_factory=list)
platforms: list[str] = field(default_factory=list)
platform_subdirs: dict[str, str] = field(default_factory=dict)
platform_system_requirements: dict[str, dict[str, str]] = field(
default_factory=dict
)
# Root-level dependency pool from [workspace.dependencies].
# Concrete feature dependencies may opt in with { workspace = true }.
workspace_dependencies: dict[str, MatchSpec] = field(default_factory=dict)
# Features keyed by name; always includes "default"
features: dict[str, Feature] = field(default_factory=dict)
# Environments keyed by name; always includes "default"
environments: dict[str, Environment] = field(default_factory=dict)
# Workspace root directory (parent of manifest file)
root: str = ""
# Path to the manifest file that was parsed
manifest_path: str = ""
# Directory for project-local environments (default: .conda/envs)
envs_dir: str = ".conda/envs"
# Preview / optional fields
channel_priority: str | None = None # "strict" | "flexible" | "disabled"
archive: ArchiveConfig = field(default_factory=ArchiveConfig)
platform_requirement_toml_aliases: ClassVar[dict[str, str]] = {
"glibc": "libc",
"__glibc": "__glibc",
"osx": "macos",
"__osx": "__osx",
"win": "windows",
"__win": "__win",
}
rich_platform_name_order: ClassVar[tuple[str, ...]] = (
"cuda",
"archspec",
"glibc",
"linux",
"osx",
"win",
)
_platform_name_segment_re: ClassVar[re.Pattern[str]] = re.compile(r"[^A-Za-z0-9]+")
@property
def _manifest_text(self) -> str | None:
"""Return the exact manifest generation accepted by the parser."""
return self.__dict__.get("_accepted_manifest_text")
@_manifest_text.setter
def _manifest_text(self, value: str | None) -> None:
self.__dict__["_accepted_manifest_text"] = value
def __post_init__(self) -> None:
"""Ensure the default feature and environment always exist.
Also validates that all declared platforms map to recognised
conda subdirs (e.g. ``linux-64``, ``osx-arm64``). Pixi rich
platforms may use workspace-scoped names such as
``linux-64-cuda``; those names stay in :attr:`platforms`, while
:attr:`platform_subdirs` records the concrete conda subdir the
solver should use.
"""
if Feature.DEFAULT_NAME not in self.features:
self.features[Feature.DEFAULT_NAME] = Feature(name=Feature.DEFAULT_NAME)
if Environment.DEFAULT_NAME not in self.environments:
self.environments[Environment.DEFAULT_NAME] = Environment(
name=Environment.DEFAULT_NAME
)
environment_keys: dict[tuple[str, ...], str] = {}
for name in self.environments:
if not is_path_segment(name):
raise EnvironmentNameInvalidError(name)
key = portable_path_key(PurePosixPath(name))
existing = environment_keys.get(key)
if existing is not None:
raise EnvironmentNameInvalidError(
name,
reason=f"it conflicts with environment '{existing}'",
)
environment_keys[key] = name
for platform in self.platforms:
self.platform_subdirs.setdefault(platform, platform)
invalid = [
subdir
for platform in self.platforms
if (subdir := self.platform_subdirs[platform]) not in KNOWN_SUBDIRS
]
if invalid:
raise PlatformError(
", ".join(invalid),
sorted(KNOWN_SUBDIRS),
)
[docs]
@staticmethod
def default_system_requirements_for_subdir(subdir: str) -> dict[str, str]:
"""Return Pixi's default rich-platform requirements for *subdir*."""
if subdir.startswith("linux-"):
return {"glibc": "2.28", "linux": "4.18"}
if subdir.startswith("osx-"):
return {"osx": "13.0"}
return {}
[docs]
def get_environment(self, name: str) -> Environment:
"""Return the environment with *name*, raising if not found."""
if name not in self.environments:
raise EnvironmentNotFoundError(name, list(self.environments.keys()))
return self.environments[name]
[docs]
def resolve_features(self, environment: Environment) -> list[Feature]:
"""Return the ordered list of features for *environment*.
By default, the ``default`` feature is prepended unless the
environment sets ``no_default_feature``.
"""
result: list[Feature] = []
if not environment.no_default_feature:
result.append(self.features[Feature.DEFAULT_NAME])
for fname in environment.features:
if fname not in self.features:
raise FeatureNotFoundError(fname, environment.name)
feat = self.features[fname]
if feat not in result:
result.append(feat)
return result
[docs]
def merged_conda_dependencies(
self,
environment: Environment,
platform: str | None = None,
) -> dict[str, MatchSpec]:
"""Merge conda dependencies across features for *environment*.
Later features override earlier ones. If *platform* is given,
target-specific dependencies are also merged in.
"""
merged: dict[str, MatchSpec] = {}
platform_keys = self.target_platform_keys(platform)
for feature in self.resolve_features(environment):
merged.update(feature.conda_dependencies)
for key in platform_keys:
if key in feature.target_conda_dependencies:
merged.update(feature.target_conda_dependencies[key])
merged.update(environment.conda_dependencies)
for key in platform_keys:
if key in environment.target_conda_dependencies:
merged.update(environment.target_conda_dependencies[key])
return merged
[docs]
def merged_pypi_dependencies(
self,
environment: Environment,
platform: str | None = None,
) -> dict[str, PyPIDependency]:
"""Merge PyPI dependencies across features for *environment*."""
merged: dict[str, PyPIDependency] = {}
platform_keys = self.target_platform_keys(platform)
for feature in self.resolve_features(environment):
merged.update(feature.pypi_dependencies)
for key in platform_keys:
if key in feature.target_pypi_dependencies:
merged.update(feature.target_pypi_dependencies[key])
merged.update(environment.pypi_dependencies)
for key in platform_keys:
if key in environment.target_pypi_dependencies:
merged.update(environment.target_pypi_dependencies[key])
return merged
[docs]
def merged_system_requirements(
self,
environment: Environment,
platform: str | None = None,
) -> dict[str, str]:
"""Merge system requirements across features for *environment*."""
merged: dict[str, str] = {}
for feature in self.resolve_features(environment):
merged.update(feature.system_requirements)
if platform and platform in self.platform_system_requirements:
merged.update(self.platform_system_requirements[platform])
return merged
[docs]
def merged_channels(self, environment: Environment) -> list[Channel]:
"""Merge channels across features for *environment*.
Feature-specific channels are appended after the workspace-level
channels, preserving priority order. Duplicates are removed.
"""
seen: set[str] = set()
result: list[Channel] = []
for ch in self.channels:
if ch.canonical_name not in seen:
seen.add(ch.canonical_name)
result.append(ch)
for feature in self.resolve_features(environment):
for ch in feature.channels:
if ch.canonical_name not in seen:
seen.add(ch.canonical_name)
result.append(ch)
return result
[docs]
@dataclass
class TaskArg:
"""A named argument that can be passed to a task."""
name: str
default: str | None = None
choices: list[str] | None = None
[docs]
def to_toml(self) -> dict[str, object]:
"""Serialize to a TOML-compatible dict."""
entry: dict[str, object] = {"arg": self.name}
if self.default is not None:
entry["default"] = self.default
if self.choices is not None:
entry["choices"] = self.choices
return entry
[docs]
@dataclass
class TaskDependency:
"""A reference to another task that must run first."""
task: str
args: list[str | dict[str, str]] = field(default_factory=list)
environment: str | None = None
[docs]
def to_toml(self) -> str | dict[str, object]:
"""Serialize to a TOML-compatible value (string or dict)."""
if self.args or self.environment:
entry: dict[str, object] = {"task": self.task}
if self.args:
entry["args"] = self.args
if self.environment:
entry["environment"] = self.environment
return entry
return self.task
[docs]
@dataclass
class TaskOverride:
"""Per-platform override for any task field.
Non-None fields replace the base task's values when the override
is merged into a Task via ``Task.resolve_for_platform``.
"""
cmd: str | list[str] | None = None
args: list[TaskArg] | None = None
depends_on: list[TaskDependency] | None = None
cwd: str | None = None
env: dict[str, str] | None = None
inputs: list[str] | None = None
outputs: list[str] | None = None
clean_env: bool | None = None
[docs]
@dataclass
class Task:
"""A single task definition with all its configuration."""
name: str
cmd: str | list[str] | None = None
args: list[TaskArg] = field(default_factory=list)
depends_on: list[TaskDependency] = field(default_factory=list)
cwd: str | None = None
env: dict[str, str] = field(default_factory=dict)
description: str | None = None
inputs: list[str] = field(default_factory=list)
outputs: list[str] = field(default_factory=list)
clean_env: bool = False
default_environment: str | None = None
platforms: dict[str, TaskOverride] | None = None
@property
def is_alias(self) -> bool:
"""True when the task is just a dependency grouping with no command."""
return self.cmd is None and bool(self.depends_on)
@property
def is_hidden(self) -> bool:
"""Hidden tasks (prefixed with ``_``) are omitted from listings."""
return self.name.startswith("_")
