Merge branch 'conda:main' into default-poetry-source-pypi

conda_lock/conda_lock.py

@@ -53,7 +53,8 @@
     determine_conda_executable,
     is_micromamba,
 )
-from conda_lock.lockfile import (
+from conda_lock.lockfile import parse_conda_lock_file, write_conda_lock_file
+from conda_lock.lockfile.v2prelim.models import (
     GitMeta,
     InputMeta,
     LockedDependency,
@@ -62,8 +63,6 @@
     MetadataOption,
     TimeMeta,
     UpdateSpecification,
-    parse_conda_lock_file,
-    write_conda_lock_file,
 )
 from conda_lock.lookup import set_lookup_location
 from conda_lock.models.channel import Channel

conda_lock/conda_solver.py

@@ -24,7 +24,8 @@
     conda_pkgs_dir,
     is_micromamba,
 )
-from conda_lock.lockfile import HashModel, LockedDependency, apply_categories
+from conda_lock.lockfile import apply_categories
+from conda_lock.lockfile.v2prelim.models import HashModel, LockedDependency
 from conda_lock.models.channel import Channel
 from conda_lock.models.lock_spec import Dependency, VersionedDependency
 

conda_lock/lockfile/__init__.py

@@ -1,26 +1,21 @@
-import json
 import pathlib
 
 from collections import defaultdict
 from textwrap import dedent
-from typing import Any, Collection, Dict, List, Mapping, Optional, Sequence, Set, Union
+from typing import Collection, Dict, List, Mapping, Optional, Sequence, Set, Union
 
 import yaml
 
+from conda_lock.lockfile.v1.models import Lockfile as LockfileV1
+from conda_lock.lockfile.v2prelim.models import (
+    LockedDependency,
+    Lockfile,
+    MetadataOption,
+    lockfile_v1_to_v2,
+)
 from conda_lock.lookup import conda_name_to_pypi_name
 from conda_lock.models.lock_spec import Dependency
 
-from .models import DependencySource as DependencySource
-from .models import GitMeta as GitMeta
-from .models import HashModel as HashModel
-from .models import InputMeta as InputMeta
-from .models import LockedDependency, Lockfile
-from .models import LockKey as LockKey
-from .models import LockMeta as LockMeta
-from .models import MetadataOption
-from .models import TimeMeta as TimeMeta
-from .models import UpdateSpecification as UpdateSpecification
-
 
 def _seperator_munge_get(
     d: Mapping[str, Union[List[LockedDependency], LockedDependency]], key: str
@@ -138,14 +133,11 @@ def parse_conda_lock_file(path: pathlib.Path) -> Lockfile:
     with path.open() as f:
         content = yaml.safe_load(f)
     version = content.pop("version", None)
-    if not (isinstance(version, int) and version <= Lockfile.version):
+    if version == 1:
+        return lockfile_v1_to_v2(LockfileV1.parse_obj(content))
+    else:
         raise ValueError(f"{path} has unknown version {version}")
 
-    for p in content["package"]:
-        del p["optional"]
-
-    return Lockfile.parse_obj(content)
-
 
 def write_conda_lock_file(
     content: Lockfile,
@@ -206,23 +198,5 @@ def write_section(text: str) -> None:
                     conda-lock {metadata_flags}{' '.join('-f '+path for path in content.metadata.sources)} --lockfile {path.name}
                 """
             )
-
-        output: Dict[str, Any] = {
-            "version": Lockfile.version,
-            "metadata": json.loads(
-                content.metadata.json(
-                    by_alias=True, exclude_unset=True, exclude_none=True
-                )
-            ),
-            "package": [
-                {
-                    **package.dict(
-                        by_alias=True, exclude_unset=True, exclude_none=True
-                    ),
-                    "optional": (package.category != "main"),
-                }
-                for package in content.package
-            ],
-        }
-
+        output = content.to_v1().dict_for_output()
         yaml.dump(output, stream=f, sort_keys=False)

conda_lock/lockfile/models.py → conda_lock/lockfile/v1/models.py

@@ -1,12 +1,22 @@
 import datetime
 import enum
 import hashlib
+import json
 import logging
 import pathlib
 import typing
 
-from collections import defaultdict, namedtuple
-from typing import TYPE_CHECKING, AbstractSet, ClassVar, Dict, List, Optional, Union
+from collections import namedtuple
+from typing import (
+    TYPE_CHECKING,
+    AbstractSet,
+    Any,
+    ClassVar,
+    Dict,
+    List,
+    Optional,
+    Union,
+)
 
 
 if TYPE_CHECKING:
@@ -36,7 +46,7 @@ class HashModel(StrictModel):
     sha256: Optional[str] = None
 
 
-class LockedDependency(StrictModel):
+class BaseLockedDependency(StrictModel):
     name: str
     version: str
     manager: Literal["conda", "pip"]
@@ -58,6 +68,10 @@ def validate_hash(cls, v: HashModel, values: Dict[str, typing.Any]) -> HashModel
         return v
 
 
+class LockedDependency(BaseLockedDependency):
+    optional: bool
+
+
 class MetadataOption(enum.Enum):
     TimeStamp = "timestamp"
     GitSha = "git_sha"
@@ -283,89 +297,15 @@ class Lockfile(StrictModel):
     package: List[LockedDependency]
     metadata: LockMeta
 
-    def __or__(self, other: "Lockfile") -> "Lockfile":
-        return other.__ror__(self)
-
-    def __ror__(self, other: "Optional[Lockfile]") -> "Lockfile":
-        """
-        merge self into other
-        """
-        if other is None:
-            return self
-        elif not isinstance(other, Lockfile):
-            raise TypeError
-
-        assert self.metadata.channels == other.metadata.channels
-
-        ours = {d.key(): d for d in self.package}
-        theirs = {d.key(): d for d in other.package}
-
-        # Pick ours preferentially
-        package: List[LockedDependency] = []
-        for key in sorted(set(ours.keys()).union(theirs.keys())):
-            if key not in ours or key[-1] not in self.metadata.platforms:
-                package.append(theirs[key])
-            else:
-                package.append(ours[key])
-
-        # Resort the conda packages topologically
-        final_package = self._toposort(package)
-        return Lockfile(package=final_package, metadata=other.metadata | self.metadata)
-
-    def toposort_inplace(self) -> None:
-        self.package = self._toposort(self.package)
-
-    @staticmethod
-    def _toposort(
-        package: List[LockedDependency], update: bool = False
-    ) -> List[LockedDependency]:
-        platforms = {d.platform for d in package}
-
-        # Resort the conda packages topologically
-        final_package: List[LockedDependency] = []
-        for platform in sorted(platforms):
-            from .._vendor.conda.common.toposort import toposort
-
-            # Add the remaining non-conda packages in the order in which they appeared.
-            # Order the pip packages topologically ordered (might be not 100% perfect if they depend on
-            # other conda packages, but good enough
-            for manager in ["conda", "pip"]:
-                lookup = defaultdict(set)
-                packages: Dict[str, LockedDependency] = {}
-
-                for d in package:
-                    if d.platform != platform:
-                        continue
-
-                    if d.manager != manager:
-                        continue
-
-                    lookup[d.name] = set(d.dependencies)
-                    packages[d.name] = d
-
-                ordered = toposort(lookup)
-                for package_name in ordered:
-                    # since we could have a pure dep in here, that does not have a package
-                    # eg a pip package that depends on a conda package (the conda package will not be in this list)
-                    dep = packages.get(package_name)
-                    if dep is None:
-                        continue
-                    if dep.manager != manager:
-                        continue
-                    # skip virtual packages
-                    if dep.manager == "conda" and dep.name.startswith("__"):
-                        continue
-
-                    final_package.append(dep)
-
-        return final_package
-
-
-class UpdateSpecification:
-    def __init__(
-        self,
-        locked: Optional[List[LockedDependency]] = None,
-        update: Optional[List[str]] = None,
-    ):
-        self.locked = locked or []
-        self.update = update or []
+    def dict_for_output(self) -> Dict[str, Any]:
+        """Convert the lockfile to a dictionary that can be written to a file."""
+        return {
+            "version": Lockfile.version,
+            "metadata": json.loads(
+                self.metadata.json(by_alias=True, exclude_unset=True, exclude_none=True)
+            ),
+            "package": [
+                package.dict(by_alias=True, exclude_unset=True, exclude_none=True)
+                for package in self.package
+            ],
+        }