Initial Version of SourceFile Approach

conda_lock/conda_lock.py

@@ -62,26 +62,20 @@
     PIP_SUPPORT = True
 except ImportError:
     PIP_SUPPORT = False
-from conda_lock.lookup import set_lookup_location
-from conda_lock.src_parser import (
-    Dependency,
+from conda_lock.lockfile import (
     GitMeta,
     InputMeta,
     LockedDependency,
     Lockfile,
     LockMeta,
-    LockSpecification,
-    MetadataOption,
     TimeMeta,
     UpdateSpecification,
-    aggregate_lock_specs,
+    parse_conda_lock_file,
+    write_conda_lock_file,
 )
-from conda_lock.src_parser.environment_yaml import parse_environment_file
-from conda_lock.src_parser.lockfile import parse_conda_lock_file, write_conda_lock_file
-from conda_lock.src_parser.meta_yaml import parse_meta_yaml_file
-from conda_lock.src_parser.pyproject_toml import parse_pyproject_toml
+from conda_lock.lookup import set_lookup_location
+from conda_lock.src_parser import LockSpecification, MetadataOption, make_lock_spec
 from conda_lock.virtual_package import (
-    FakeRepoData,
     default_virtual_package_repodata,
     virtual_package_repo_from_specification,
 )
@@ -114,8 +108,6 @@
     sys.exit(1)
 
 
-DEFAULT_PLATFORMS = ["osx-64", "linux-64", "win-64"]
-
 KIND_EXPLICIT: Literal["explicit"] = "explicit"
 KIND_LOCK: Literal["lock"] = "lock"
 KIND_ENV: Literal["env"] = "env"
@@ -243,44 +235,6 @@ def fn_to_dist_name(fn: str) -> str:
     return fn
 
 
-def make_lock_spec(
-    *,
-    src_files: List[pathlib.Path],
-    virtual_package_repo: FakeRepoData,
-    channel_overrides: Optional[Sequence[str]] = None,
-    platform_overrides: Optional[Sequence[str]] = None,
-    required_categories: Optional[AbstractSet[str]] = None,
-) -> LockSpecification:
-    """Generate the lockfile specs from a set of input src_files.  If required_categories is set filter out specs that do not match those"""
-    lock_specs = parse_source_files(
-        src_files=src_files, platform_overrides=platform_overrides
-    )
-
-    lock_spec = aggregate_lock_specs(lock_specs)
-    lock_spec.virtual_package_repo = virtual_package_repo
-    lock_spec.channels = (
-        [Channel.from_string(co) for co in channel_overrides]
-        if channel_overrides
-        else lock_spec.channels
-    )
-    lock_spec.platforms = (
-        list(platform_overrides) if platform_overrides else lock_spec.platforms
-    ) or list(DEFAULT_PLATFORMS)
-
-    if required_categories is not None:
-
-        def dep_has_category(d: Dependency, categories: AbstractSet[str]) -> bool:
-            return d.category in categories
-
-        lock_spec.dependencies = [
-            d
-            for d in lock_spec.dependencies
-            if dep_has_category(d, categories=required_categories)
-        ]
-
-    return lock_spec
-
-
 def make_lock_files(
     *,
     conda: PathLike,
@@ -353,11 +307,12 @@ def make_lock_files(
 
     with virtual_package_repo:
         lock_spec = make_lock_spec(
-            src_files=src_files,
+            src_file_paths=src_files,
             channel_overrides=channel_overrides,
-            platform_overrides=platform_overrides,
+            platform_overrides=set(platform_overrides) if platform_overrides else set(),
             virtual_package_repo=virtual_package_repo,
             required_categories=required_categories if filter_categories else None,
+            pip_support=PIP_SUPPORT,
         )
         lock_content: Optional[Lockfile] = None
 
@@ -704,12 +659,8 @@ def _solve_for_arch(
     """
     if update_spec is None:
         update_spec = UpdateSpecification()
-    # filter requested and locked dependencies to the current platform
-    dependencies = [
-        dep
-        for dep in spec.dependencies
-        if (not dep.selectors.platform) or platform in dep.selectors.platform
-    ]
+    dependencies = spec.dependencies[platform]
+
     locked = [dep for dep in update_spec.locked if dep.platform == platform]
     requested_deps_by_name = {
         manager: {dep.name: dep for dep in dependencies if dep.manager == manager}
@@ -867,42 +818,6 @@ def create_lockfile_from_spec(
     )
 
 
-def parse_source_files(
-    src_files: List[pathlib.Path],
-    platform_overrides: Optional[Sequence[str]],
-) -> List[LockSpecification]:
-    """
-    Parse a sequence of dependency specifications from source files
-
-    Parameters
-    ----------
-    src_files :
-        Files to parse for dependencies
-    platform_overrides :
-        Target platforms to render environment.yaml and meta.yaml files for
-    """
-    desired_envs: List[LockSpecification] = []
-    for src_file in src_files:
-        if src_file.name == "meta.yaml":
-            desired_envs.append(
-                parse_meta_yaml_file(
-                    src_file, list(platform_overrides or DEFAULT_PLATFORMS)
-                )
-            )
-        elif src_file.name == "pyproject.toml":
-            desired_envs.append(parse_pyproject_toml(src_file))
-        else:
-            desired_envs.append(
-                parse_environment_file(
-                    src_file,
-                    platform_overrides,
-                    default_platforms=DEFAULT_PLATFORMS,
-                    pip_support=PIP_SUPPORT,
-                )
-            )
-    return desired_envs
-
-
 def _add_auth_to_line(line: str, auth: Dict[str, str]) -> str:
     matching_auths = [a for a in auth if a in line]
     if not matching_auths:

conda_lock/conda_solver.py

@@ -31,14 +31,9 @@
     conda_pkgs_dir,
     is_micromamba,
 )
+from conda_lock.lockfile import HashModel, LockedDependency, _apply_categories
 from conda_lock.models.channel import Channel
-from conda_lock.src_parser import (
-    Dependency,
-    HashModel,
-    LockedDependency,
-    VersionedDependency,
-    _apply_categories,
-)
+from conda_lock.src_parser import Dependency, VersionedDependency
 
 
 logger = logging.getLogger(__name__)

conda_lock/src_parser/lockfile.py → conda_lock/lockfile/__init__.py

@@ -1,14 +1,79 @@
 import json
 import pathlib
 
+from collections import defaultdict
 from textwrap import dedent
-from typing import Collection, Optional
+from typing import Collection, Dict, List, Optional, Sequence, Set
 
 import yaml
 
-from conda_lock.src_parser import MetadataOption
+from conda_lock.src_parser import Dependency, MetadataOption
 
-from . import Lockfile
+from .models import *
+from .models import LockedDependency, Lockfile
+
+
+def _apply_categories(
+    requested: Dict[str, Dependency],
+    planned: Dict[str, LockedDependency],
+    categories: Sequence[str] = ("main", "dev"),
+) -> None:
+    """map each package onto the root request the with the highest-priority category"""
+    # walk dependency tree to assemble all transitive dependencies by request
+    dependents: Dict[str, Set[str]] = {}
+    by_category = defaultdict(list)
+
+    def seperator_munge_get(
+        d: Dict[str, LockedDependency], key: str
+    ) -> LockedDependency:
+        # since separators are not consistent across managers (or even within) we need to do some double attempts here
+        try:
+            return d[key]
+        except KeyError:
+            try:
+                return d[key.replace("-", "_")]
+            except KeyError:
+                return d[key.replace("_", "-")]
+
+    for name, request in requested.items():
+        todo: List[str] = list()
+        deps: Set[str] = set()
+        item = name
+        while True:
+            todo.extend(
+                dep
+                for dep in seperator_munge_get(planned, item).dependencies
+                # exclude virtual packages
+                if not (dep in deps or dep.startswith("__"))
+            )
+            if todo:
+                item = todo.pop(0)
+                deps.add(item)
+            else:
+                break
+
+        dependents[name] = deps
+
+        by_category[request.category].append(request.name)
+
+    # now, map each package to its root request
+    categories = [*categories, *(k for k in by_category if k not in categories)]
+    root_requests = {}
+    for category in categories:
+        for root in by_category.get(category, []):
+            for transitive_dep in dependents[root]:
+                if transitive_dep not in root_requests:
+                    root_requests[transitive_dep] = root
+    # include root requests themselves
+    for name in requested:
+        root_requests[name] = name
+
+    for dep, root in root_requests.items():
+        source = requested[root]
+        # try a conda target first
+        target = seperator_munge_get(planned, dep)
+        target.category = source.category
+        target.optional = source.optional
 
 
 def parse_conda_lock_file(