Skip to content

Commit

Permalink
Add a very rough export of lock spec to pixi.toml
Browse files Browse the repository at this point in the history
  • Loading branch information
maresb committed Sep 7, 2024
1 parent b7b36c4 commit a830b90
Show file tree
Hide file tree
Showing 2 changed files with 212 additions and 1 deletion.
6 changes: 6 additions & 0 deletions conda_lock/conda_lock.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,11 @@ def make_lock_files( # noqa: C901
platform_overrides=platform_overrides,
required_categories=required_categories if filter_categories else None,
)
from conda_lock.models.lock_spec import render_pixi_toml

pixi_toml = render_pixi_toml(lock_spec=lock_spec)
print("\n".join(pixi_toml))
sys.exit(0)

# Load existing lockfile if it exists
original_lock_content: Optional[Lockfile] = None
Expand Down Expand Up @@ -1619,6 +1624,7 @@ def render(
# bail out if we do not encounter the lockfile
lock_file = pathlib.Path(lock_file)
if not lock_file.exists():
print(f"ERROR: Lockfile {lock_file} does not exist.\n\n", file=sys.stderr)
print(ctx.get_help())
sys.exit(1)

Expand Down
207 changes: 206 additions & 1 deletion conda_lock/models/lock_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,10 @@
import json
import pathlib
import typing
import warnings

from typing import Dict, List, Optional, Union
from collections import defaultdict
from typing import Any, Dict, List, NamedTuple, Optional, Set, Union

from pydantic import BaseModel, Field, validator
from typing_extensions import Literal
Expand Down Expand Up @@ -120,3 +122,206 @@ def validate_pip_repositories(
if isinstance(repository, str):
value[index] = PipRepository.from_string(repository)
return typing.cast(List[PipRepository], value)


class DepKey1(NamedTuple):
name: str
category: str
platform: str
manager: str


class DepKey2(NamedTuple):
name: str
category: str
manager: str


class DepWithPlatform(NamedTuple):
dep: Dependency
platform: str


class DepWithSubdir(NamedTuple):
dep: Dependency
subdir: Optional[str]


def render_pixi_toml( # noqa: C901
*,
lock_spec: LockSpecification,
project_name: str = "project-name-placeholder",
) -> List[str]:
all_platforms = lock_spec.dependencies.keys()

all_categories: Set[str] = set()
for platform in all_platforms:
for dep in lock_spec.dependencies[platform]:
all_categories.add(dep.category)
if {"main", "default"} <= all_categories:
raise ValueError("Cannot have both 'main' and 'default' as categories/extras")

indexed_deps: Dict[DepKey1, Dependency] = {}
for platform in all_platforms:
deps = lock_spec.dependencies[platform]
for dep in deps:
category = dep.category
key1 = DepKey1(
name=dep.name, category=category, platform=platform, manager=dep.manager
)
if key1 in indexed_deps:
raise ValueError(
f"Duplicate dependency {key1}: {dep}, {indexed_deps[key1]}"
)
indexed_deps[key1] = dep

# Collect by platform
aggregated_deps: Dict[DepKey2, List[DepWithPlatform]] = defaultdict(list)
for key1, dep in indexed_deps.items():
key2 = DepKey2(name=key1.name, category=key1.category, manager=key1.manager)
aggregated_deps[key2].append(DepWithPlatform(dep=dep, platform=key1.platform))

# Reduce by platform
reduced_deps: Dict[DepKey2, DepWithSubdir] = {}
for key2, deps_with_platforms in aggregated_deps.items():
for curr, next in zip(deps_with_platforms, deps_with_platforms[1:]):
if curr.dep != next.dep:
raise ValueError(f"Conflicting dependencies {curr} and {next}")
dep = deps_with_platforms[0].dep
dep_platforms = {dep.platform for dep in deps_with_platforms}
if len(dep_platforms) < len(all_platforms):
if len(dep_platforms) != 1:
raise ValueError(
f"Dependency {dep} is specified for more than one platform but not all: {dep_platforms=}, {all_platforms=}"
)
reduced_deps[key2] = DepWithSubdir(dep=dep, subdir=dep_platforms.pop())
elif len(dep_platforms) == len(all_platforms):
reduced_deps[key2] = DepWithSubdir(dep=dep, subdir=None)
else:
raise RuntimeError(f"Impossible: {dep_platforms=}, {all_platforms=}")

result: List[str] = [
"# This file was generated by conda-lock for the pixi environment manager.",
"# For more information, see <https://pixi.sh>." "# Source files:",
]
for source in lock_spec.sources:
result.append(f"# - {source}")
result.extend(
[
"",
"[project]",
f'name = "{project_name}"',
f"platforms = {list(all_platforms)}",
]
)
channels: List[str] = [channel.url for channel in lock_spec.channels]
for channel in lock_spec.channels:
if channel.used_env_vars:
warnings.warn(
f"Channel {channel.url} uses environment variables "
"which are not implemented"
)
result.append(f"channels = {channels}")
result.append("")

sorted_categories = sorted(all_categories)
if "main" in sorted_categories:
sorted_categories.remove("main")
sorted_categories.insert(0, "main")
if "default" in sorted_categories:
sorted_categories.remove("default")
sorted_categories.insert(0, "default")

for category in sorted_categories:
feature = category if category != "main" else "default"

conda_deps: Dict[str, DepWithSubdir] = {}
pip_deps: Dict[str, Dependency] = {}
for key2, dep_with_subdir in sorted(
reduced_deps.items(), key=lambda x: x[0].name
):
if key2.category == category:
name = key2.name
manager = dep_with_subdir.dep.manager
if manager == "conda":
conda_deps[name] = dep_with_subdir
elif manager == "pip":
if dep_with_subdir.subdir is not None:
raise ValueError(
f"Subdir specified for pip dependency {dep_with_subdir}"
)
else:
pip_deps[name] = dep_with_subdir.dep
else:
raise ValueError(f"Unknown manager {manager}")

if conda_deps:
if feature == "default":
result.append("[dependencies]")
else:
result.append(f"[feature.{feature}.dependencies]")
for name, dep_with_subdir in conda_deps.items():
pixi_spec = make_pixi_conda_spec(dep_with_subdir)
result.append(f"{name} = {pixi_spec}")
result.append("")

if pip_deps:
if feature == "default":
result.append("[pypi-dependencies]")
else:
result.append(f"[feature.{feature}.pypi-dependencies]")
for name, dep in pip_deps.items():
pixi_spec = make_pixi_pip_spec(dep)
result.append(f"{name} = {pixi_spec}")
result.append("")
return result


def make_pixi_conda_spec(dep_with_subdir: DepWithSubdir) -> str:
dep = dep_with_subdir.dep
subdir = dep_with_subdir.subdir
matchspec = {}
if dep.extras:
warnings.warn(f"Extras not supported in Conda dep {dep}")
if isinstance(dep, VersionedDependency):
matchspec["version"] = dep.version or "*"
if subdir is not None:
matchspec["subdir"] = subdir
if dep.build is not None:
matchspec["build"] = dep.build
if dep.hash is not None:
raise NotImplementedError(f"Hash not yet supported in {dep}")
if dep.conda_channel is not None:
matchspec["channel"] = dep.conda_channel
if len(matchspec) == 1:
return f'"{matchspec["version"]}"'
else:
return json.dumps(matchspec)
elif isinstance(dep, URLDependency):
raise NotImplementedError(f"URL not yet supported in {dep}")
elif isinstance(dep, VCSDependency):
raise NotImplementedError(f"VCS not yet supported in {dep}")
else:
raise ValueError(f"Unknown dependency type {dep}")


def make_pixi_pip_spec(dep: Dependency) -> str:
matchspec: Dict[str, Any] = {}
if isinstance(dep, VersionedDependency):
matchspec["version"] = dep.version or "*"
if dep.hash is not None:
raise NotImplementedError(f"Hash not yet supported in {dep}")
if dep.conda_channel is not None:
matchspec["channel"] = dep.conda_channel
if dep.extras:
matchspec["extras"] = dep.extras
if len(matchspec) == 1:
return f'"{matchspec["version"]}"'
else:
return json.dumps(matchspec)
elif isinstance(dep, URLDependency):
raise NotImplementedError(f"URL not yet supported in {dep}")
elif isinstance(dep, VCSDependency):
raise NotImplementedError(f"VCS not yet supported in {dep}")
else:
raise ValueError(f"Unknown dependency type {dep}")

0 comments on commit a830b90

Please sign in to comment.