"""Manifest detection and parser registry (workspaces and tasks).
Search order (same for workspaces and tasks)
--------------------------------------------
1. ``conda.toml`` -- conda-native manifest format
2. ``pixi.toml`` -- pixi-native format (compatibility)
3. ``pyproject.toml`` -- pixi or conda tables embedded
The first file that exists *and* contains the relevant configuration wins.
"""
from __future__ import annotations
import os
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING
from ..exceptions import (
NoTaskFileError,
WorkspaceNotFoundError,
WorkspaceParseError,
)
from ..models import redact_url_text
from ..paths import validate_path_parent
from .base import ManifestParser
from .pixi_toml import PixiTomlParser
from .pyproject_toml import PyprojectTomlParser
from .toml import CondaTomlParser
if TYPE_CHECKING:
from ..models import Task, WorkspaceConfig
_PARSERS: list[ManifestParser] = [
CondaTomlParser(),
PixiTomlParser(),
PyprojectTomlParser(),
]
_SEARCH_FILES: list[str] = [
"conda.toml",
"pixi.toml",
"pyproject.toml",
]
PARSER_BY_FILENAME: dict[str, ManifestParser] = {
fname: parser for parser in _PARSERS for fname in parser.filenames
}
[docs]
def walk_manifests(
start_dir: Path,
predicate: str,
*,
reject_symlinks: bool = False,
) -> Path | None:
"""Walk up from *start_dir* looking for a manifest matching *predicate*.
*predicate* is a method name on ``ManifestParser`` — either
``"has_workspace"`` or ``"has_tasks"``. Returns the first
matching file path, or ``None`` if none is found.
"""
current = start_dir.resolve()
while True:
for fname in _SEARCH_FILES:
candidate = current / fname
if not candidate.is_file():
continue
parser = PARSER_BY_FILENAME.get(fname)
if parser is None:
continue
if reject_symlinks and candidate.is_symlink():
try:
selected = getattr(parser, predicate)(candidate)
except Exception:
continue
if selected:
raise WorkspaceParseError(
candidate,
"symbolic links are not supported for this operation",
)
continue
try:
selected = getattr(parser, predicate)(candidate)
except WorkspaceParseError:
raise
except Exception as exc:
raise WorkspaceParseError(candidate, redact_url_text(str(exc))) from exc
if selected:
return candidate
parent = current.parent
if parent == current:
break
current = parent
return None
[docs]
def detect_workspace_file(
start_dir: str | Path | None = None,
*,
reject_symlinks: bool = False,
) -> Path:
"""Walk up from *start_dir* to find a workspace manifest.
Returns the path to the first matching file.
Raises ``WorkspaceNotFoundError`` if none is found.
"""
if start_dir is None:
start_dir = Path.cwd()
else:
start_dir = Path(start_dir)
result = walk_manifests(
start_dir,
"has_workspace",
reject_symlinks=reject_symlinks,
)
if result is None:
raise WorkspaceNotFoundError(start_dir)
return result
[docs]
def find_parser(path: Path) -> ManifestParser:
"""Return the parser that can handle *path*.
Raises ``WorkspaceParseError`` if no parser matches.
"""
for parser in _PARSERS:
if parser.can_handle(path):
return parser
raise WorkspaceParseError(path, f"No parser available for '{path.name}'")
[docs]
def detect_and_parse(
source: str | Path | None = None,
) -> tuple[Path, WorkspaceConfig]:
"""Resolve *source* to a workspace manifest and parse it.
Returns ``(manifest_path, workspace_config)``.
"""
path = ManifestParser.resolve_source(Path(source or Path.cwd()))
try:
manifest_text = ManifestParser.read_manifest_text(path)
except WorkspaceParseError:
raise
except Exception as exc:
raise WorkspaceParseError(path, redact_url_text(str(exc))) from exc
config = find_parser(path).parse_text(path, manifest_text)
return path, config
[docs]
def detect_task_file(
start_dir: Path | None = None,
*,
reject_symlinks: bool = False,
) -> Path | None:
"""Walk up from *start_dir* looking for a file that contains tasks.
Returns the first match according to ``_SEARCH_FILES``, or ``None``.
"""
if start_dir is None:
start_dir = Path.cwd()
return walk_manifests(
Path(start_dir),
"has_tasks",
reject_symlinks=reject_symlinks,
)
[docs]
@lru_cache(maxsize=4)
def cached_task_parse(path_str: str) -> dict[str, Task]:
"""Parse tasks from a manifest file (cached by path string)."""
p = Path(path_str)
parser = find_parser(p)
return parser.parse_tasks(p)
[docs]
@lru_cache(maxsize=1)
def cached_user_task_parse(path_str: str) -> dict[str, Task]:
"""Parse tasks from the user-level ``tasks.toml`` (conda.toml format)."""
return CondaTomlParser().parse_tasks(Path(path_str))
[docs]
def user_task_file() -> Path | None:
"""Return the user-level task file path, or ``None``."""
candidates: list[Path] = []
xdg = os.environ.get("XDG_CONFIG_HOME", "")
if xdg:
candidates.append(Path(xdg) / "conda" / "tasks.toml")
candidates.append(Path.home() / ".config" / "conda" / "tasks.toml")
candidates.append(Path.home() / ".conda" / "tasks.toml")
for c in candidates:
if c.is_file():
return c
return None
[docs]
def detect_and_parse_tasks(
file_path: Path | None = None,
start_dir: Path | None = None,
*,
reject_symlinks: bool = False,
) -> tuple[Path, dict[str, Task], set[str]]:
"""Detect task files and parse them, merging user-level tasks.
Returns ``(resolved_path, {task_name: Task}, user_only_names)`` where
*user_only_names* is the set of task names that came from the user-level
file and were not overridden by the project.
Raises ``NoTaskFileError`` when neither a project nor a user task file
is found.
"""
project_path: Path | None = None
if file_path is not None:
if reject_symlinks and file_path.is_symlink():
raise WorkspaceParseError(
file_path,
"symbolic links are not supported for this operation",
)
if reject_symlinks:
project_path = file_path.absolute()
validate_path_parent(project_path)
else:
project_path = file_path.resolve()
else:
project_path = detect_task_file(
start_dir,
reject_symlinks=reject_symlinks,
)
user_path = user_task_file()
if project_path is None and user_path is None:
raise NoTaskFileError(str(start_dir or Path.cwd()))
user_tasks: dict[str, Task] = {}
if user_path is not None:
user_tasks = cached_user_task_parse(str(user_path))
if project_path is not None:
project_tasks = cached_task_parse(str(project_path))
merged = {**user_tasks, **project_tasks}
user_only = set(user_tasks) - set(project_tasks)
return project_path, merged, user_only
assert user_path is not None
return user_path, dict(user_tasks), set(user_tasks)
