feat: add pypi-to-conda-name overrides to pyproject parsing

conda_lock/lookup.py

@@ -1,25 +1,42 @@
-from collections import ChainMap
+import logging
+
 from contextlib import suppress
 from functools import cached_property
-from typing import Dict, Mapping, Optional, Union, cast
+from typing import ClassVar, Dict, Optional, Union, cast
 
 import requests
 import yaml
 
 from packaging.utils import NormalizedName, canonicalize_name
-from typing_extensions import NotRequired, TypedDict
+from typing_extensions import TypedDict
+
+
+DEFAULT_MAPPING_URL = "https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/mappings/pypi/grayskull_pypi_mapping.yaml"
 
 
 class MappingEntry(TypedDict):
     conda_name: str
-    # legacy field, generally not used by anything anymore
-    conda_forge: NotRequired[str]
     pypi_name: NormalizedName
 
 
 class _LookupLoader:
-    _mapping_url: str = "https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/mappings/pypi/grayskull_pypi_mapping.yaml"
-    _local_mappings: Optional[Dict[NormalizedName, MappingEntry]] = None
+    """Object used to map PyPI package names to conda names."""
+
+    _SINGLETON: ClassVar[Optional["_LookupLoader"]] = None
+
+    @classmethod
+    def instance(cls) -> "_LookupLoader":
+        if cls._SINGLETON is None:
+            cls._SINGLETON = cls()
+        return cls._SINGLETON
+
+    def __init__(
+        self,
+        pypi_lookup_overrides: Optional[Dict[NormalizedName, MappingEntry]] = None,
+        mapping_url: str = DEFAULT_MAPPING_URL,
+    ) -> None:
+        self._mapping_url = mapping_url
+        self._local_mappings = pypi_lookup_overrides
 
     @property
     def mapping_url(self) -> str:
@@ -30,6 +47,8 @@ def mapping_url(self, value: str) -> None:
         # these will raise AttributeError if they haven't been cached yet.
         with suppress(AttributeError):
             del self.remote_mappings
+        with suppress(AttributeError):
+            del self.pypi_lookup
         with suppress(AttributeError):
             del self.conda_lookup
         self._mapping_url = value
@@ -53,13 +72,13 @@ def local_mappings(self) -> Dict[NormalizedName, MappingEntry]:
         return self._local_mappings or {}
 
     @local_mappings.setter
-    def local_mappings(self, mappings: Mapping[str, Union[str, MappingEntry]]) -> None:
+    def local_mappings(self, mappings: Dict[str, Union[str, MappingEntry]]) -> None:
         """Value should be a mapping from pypi name to conda name or a mapping entry."""
         lookup: Dict[NormalizedName, MappingEntry] = {}
         # normalize to Dict[NormalizedName, MappingEntry]
         for k, v in mappings.items():
             key = canonicalize_name(k)
-            if isinstance(v, Mapping):
+            if isinstance(v, dict):
                 if "conda_name" not in v or "pypi_name" not in v:
                     raise ValueError(
                         "MappingEntries must have both a 'conda_name' and 'pypi_name'"
@@ -69,61 +88,49 @@ def local_mappings(self, mappings: Mapping[str, Union[str, MappingEntry]]) -> No
             elif isinstance(v, str):
                 entry = {"conda_name": v, "pypi_name": key}
             else:
-                raise TypeError(
-                    "Each entry in the mapping must be a string or a mapping"
-                )
+                raise TypeError("Each entry in the mapping must be a string or a dict")
             lookup[key] = entry
         self._local_mappings = lookup
 
-    @property
-    def pypi_lookup(self) -> Mapping[NormalizedName, MappingEntry]:
-        """ChainMap of PyPI to conda name mappings.
+    @cached_property
+    def pypi_lookup(self) -> Dict[NormalizedName, MappingEntry]:
+        """Dict of PyPI to conda name mappings.
 
         Local mappings take precedence over remote mappings fetched from `_mapping_url`.
         """
-        return ChainMap(self.local_mappings, self.remote_mappings)
+        return {**self.remote_mappings, **self.local_mappings}
 
     @cached_property
     def conda_lookup(self) -> Dict[str, MappingEntry]:
         return {record["conda_name"]: record for record in self.pypi_lookup.values()}
 
 
-LOOKUP_OBJECT = _LookupLoader()
-
-
-def get_forward_lookup() -> Mapping[NormalizedName, MappingEntry]:
-    global LOOKUP_OBJECT
-    return LOOKUP_OBJECT.pypi_lookup
-
+def set_lookup_location(lookup_url: str) -> None:
+    """Set the location of the pypi lookup
 
-def get_lookup() -> Dict[str, MappingEntry]:
+    Used by the `lock` cli command to override the DEFAULT_MAPPING_URL for the lookup.
     """
-    Reverse grayskull name mapping to map conda names onto PyPI
-    """
-    global LOOKUP_OBJECT
-    return LOOKUP_OBJECT.conda_lookup
-
-
-def set_lookup_location(lookup_url: str) -> None:
-    global LOOKUP_OBJECT
-    LOOKUP_OBJECT.mapping_url = lookup_url
+    _LookupLoader.instance().mapping_url = lookup_url
 
 
-def set_pypi_lookup_overrides(mappings: Mapping[str, Union[str, MappingEntry]]) -> None:
+def set_pypi_lookup_overrides(mappings: Dict[str, Union[str, MappingEntry]]) -> None:
     """Set overrides to the pypi lookup"""
-    global LOOKUP_OBJECT
     # type ignore because the setter will normalize the types
-    LOOKUP_OBJECT.local_mappings = mappings  # type: ignore [assignment]
+    _LookupLoader.instance().local_mappings = mappings  # type: ignore [assignment]
 
 
 def conda_name_to_pypi_name(name: str) -> NormalizedName:
     """return the pypi name for a conda package"""
-    lookup = get_lookup()
+    lookup = _LookupLoader.instance().conda_lookup
     cname = canonicalize_name(name)
     return lookup.get(cname, {"pypi_name": cname})["pypi_name"]
 
 
 def pypi_name_to_conda_name(name: str) -> str:
     """return the conda name for a pypi package"""
     cname = canonicalize_name(name)
-    return get_forward_lookup().get(cname, {"conda_name": cname})["conda_name"]
+    forward_lookup = _LookupLoader.instance().pypi_lookup
+    if cname not in forward_lookup:
+        logging.warning(f"Could not find conda name for {cname!r}. Assuming identity.")
+        return cname
+    return forward_lookup[cname]["conda_name"]

conda_lock/src_parser/pyproject_toml.py

@@ -30,8 +30,7 @@
 from typing_extensions import Literal
 
 from conda_lock.common import get_in
-from conda_lock.lookup import get_forward_lookup as get_lookup
-from conda_lock.lookup import set_pypi_lookup_overrides
+from conda_lock.lookup import pypi_name_to_conda_name, set_pypi_lookup_overrides
 from conda_lock.models.lock_spec import (
     Dependency,
     LockSpecification,
@@ -74,22 +73,6 @@ def join_version_components(pieces: Sequence[Union[str, int]]) -> str:
     return ".".join(str(p) for p in pieces)
 
 
-def normalize_pypi_name(name: str) -> str:
-    cname = canonicalize_pypi_name(name)
-    if cname in get_lookup():
-        lookup = get_lookup()[cname]
-        res = lookup.get("conda_name") or lookup.get("conda_forge")
-        if res is not None:
-            return res
-        else:
-            logging.warning(
-                f"Could not find conda name for {cname}. Assuming identity."
-            )
-            return cname
-    else:
-        return cname
-
-
 def poetry_version_to_conda_version(version_string: Optional[str]) -> Optional[str]:
     if version_string is None:
         return None
@@ -276,7 +259,7 @@ def parse_poetry_pyproject_toml(
                 )
 
             if manager == "conda":
-                name = normalize_pypi_name(depname)
+                name = pypi_name_to_conda_name(depname)
                 version = poetry_version_to_conda_version(poetry_version_spec)
             else:
                 name = depname
@@ -422,16 +405,15 @@ def parse_python_requirement(
 ) -> Dependency:
     """Parse a requirements.txt like requirement to a conda spec"""
     parsed_req = parse_requirement_specifier(requirement)
-    name = canonicalize_pypi_name(parsed_req.name)
     collapsed_version = str(parsed_req.specifier)
     conda_version = poetry_version_to_conda_version(collapsed_version)
     if conda_version:
         conda_version = ",".join(sorted(conda_version.split(",")))
 
     if normalize_name:
-        conda_dep_name = normalize_pypi_name(name)
+        conda_dep_name = pypi_name_to_conda_name(parsed_req.name)
     else:
-        conda_dep_name = name
+        conda_dep_name = canonicalize_pypi_name(parsed_req.name)
     extras = list(parsed_req.extras)
 
     if parsed_req.url and parsed_req.url.startswith("git+"):