Move dependency checking logic to CLI (#1582)

backend/src/hatchling/cli/dep/__init__.py

@@ -13,7 +13,7 @@ def synced_impl(*, dependencies: list[str], python: str) -> None:
 
     from packaging.requirements import Requirement
 
-    from hatchling.dep.core import dependencies_in_sync
+    from hatchling.cli.dep.core import dependencies_in_sync
 
     sys_path = None
     if python:

pyproject.toml

@@ -112,6 +112,7 @@ source_pkgs = ["hatch", "hatchling", "tests"]
 omit = [
   "backend/src/hatchling/__main__.py",
   "backend/src/hatchling/bridge/*",
+  "backend/src/hatchling/cli/dep/*",
   "backend/src/hatchling/ouroboros.py",
   "src/hatch/__main__.py",
   "src/hatch/cli/new/migrate.py",

src/hatch/cli/application.py

@@ -243,8 +243,8 @@ def ensure_plugin_dependencies(self, dependencies: list[Requirement], *, wait_me
         if not dependencies:
             return
 
+        from hatch.dep.sync import dependencies_in_sync
         from hatch.env.utils import add_verbosity_flag
-        from hatchling.dep.core import dependencies_in_sync
 
         if app_path := os.environ.get('PYAPP'):
             from hatch.utils.env import PythonInfo

src/hatch/cli/project/metadata.py

@@ -18,7 +18,7 @@ def metadata(app, field):
     """
     import json
 
-    from hatchling.dep.core import dependencies_in_sync
+    from hatch.dep.sync import dependencies_in_sync
 
     if dependencies_in_sync(app.project.metadata.build.requires_complex):
         from hatchling.metadata.utils import resolve_metadata_fields

src/hatch/cli/version/__init__.py

@@ -26,7 +26,7 @@ def version(app: Application, desired_version: str | None):
             app.display(app.project.metadata.config['project']['version'])
             return
 
-    from hatchling.dep.core import dependencies_in_sync
+    from hatch.dep.sync import dependencies_in_sync
 
     with app.project.location.as_cwd():
         if not (

src/hatch/dep/sync.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import re
+import sys
+from importlib.metadata import Distribution, DistributionFinder
+
+from packaging.markers import default_environment
+from packaging.requirements import Requirement
+
+
+class DistributionCache:
+    def __init__(self, sys_path: list[str]) -> None:
+        self._resolver = Distribution.discover(context=DistributionFinder.Context(path=sys_path))
+        self._distributions: dict[str, Distribution] = {}
+        self._search_exhausted = False
+        self._canonical_regex = re.compile(r'[-_.]+')
+
+    def __getitem__(self, item: str) -> Distribution | None:
+        item = self._canonical_regex.sub('-', item).lower()
+        possible_distribution = self._distributions.get(item)
+        if possible_distribution is not None:
+            return possible_distribution
+
+        # Be safe even though the code as-is will never reach this since
+        # the first unknown distribution will fail fast
+        if self._search_exhausted:  # no cov
+            return None
+
+        for distribution in self._resolver:
+            name = distribution.metadata['Name']
+            if name is None:
+                continue
+
+            name = self._canonical_regex.sub('-', name).lower()
+            self._distributions[name] = distribution
+            if name == item:
+                return distribution
+
+        self._search_exhausted = True
+
+        return None
+
+
+def dependency_in_sync(
+    requirement: Requirement, environment: dict[str, str], installed_distributions: DistributionCache
+) -> bool:
+    if requirement.marker and not requirement.marker.evaluate(environment):
+        return True
+
+    distribution = installed_distributions[requirement.name]
+    if distribution is None:
+        return False
+
+    extras = requirement.extras
+    if extras:
+        transitive_requirements: list[str] = distribution.metadata.get_all('Requires-Dist', [])
+        if not transitive_requirements:
+            return False
+
+        available_extras: list[str] = distribution.metadata.get_all('Provides-Extra', [])
+
+        for requirement_string in transitive_requirements:
+            transitive_requirement = Requirement(requirement_string)
+            if not transitive_requirement.marker:
+                continue
+
+            for extra in extras:
+                # FIXME: This may cause a build to never be ready if newer versions do not provide the desired
+                # extra and it's just a user error/typo. See: https://github.com/pypa/pip/issues/7122
+                if extra not in available_extras:
+                    return False
+
+                extra_environment = dict(environment)
+                extra_environment['extra'] = extra
+                if not dependency_in_sync(transitive_requirement, extra_environment, installed_distributions):
+                    return False
+
+    if requirement.specifier and not requirement.specifier.contains(distribution.version):
+        return False
+
+    # TODO: handle https://discuss.python.org/t/11938
+    if requirement.url:
+        direct_url_file = distribution.read_text('direct_url.json')
+        if direct_url_file is not None:
+            import json
+
+            # https://packaging.python.org/specifications/direct-url/
+            direct_url_data = json.loads(direct_url_file)
+            if 'vcs_info' in direct_url_data:
+                url = direct_url_data['url']
+                vcs_info = direct_url_data['vcs_info']
+                vcs = vcs_info['vcs']
+                commit_id = vcs_info['commit_id']
+                requested_revision = vcs_info.get('requested_revision')
+
+                # Try a few variations, see https://peps.python.org/pep-0440/#direct-references
+                if (
+                    requested_revision and requirement.url == f'{vcs}+{url}@{requested_revision}#{commit_id}'
+                ) or requirement.url == f'{vcs}+{url}@{commit_id}':
+                    return True
+
+                if requirement.url in {f'{vcs}+{url}', f'{vcs}+{url}@{requested_revision}'}:
+                    import subprocess
+
+                    if vcs == 'git':
+                        vcs_cmd = [vcs, 'ls-remote', url]
+                        if requested_revision:
+                            vcs_cmd.append(requested_revision)
+                    # TODO: add elifs for hg, svn, and bzr https://github.com/pypa/hatch/issues/760
+                    else:
+                        return False
+                    result = subprocess.run(vcs_cmd, capture_output=True, text=True)  # noqa: PLW1510
+                    if result.returncode or not result.stdout.strip():
+                        return False
+                    latest_commit_id, *_ = result.stdout.split()
+                    return commit_id == latest_commit_id
+
+                return False
+
+    return True
+
+
+def dependencies_in_sync(
+    requirements: list[Requirement], sys_path: list[str] | None = None, environment: dict[str, str] | None = None
+) -> bool:
+    if sys_path is None:
+        sys_path = sys.path
+    if environment is None:
+        environment = default_environment()  # type: ignore
+
+    installed_distributions = DistributionCache(sys_path)
+    return all(dependency_in_sync(requirement, environment, installed_distributions) for requirement in requirements)  # type: ignore

src/hatch/env/system.py

@@ -37,7 +37,7 @@ def dependencies_in_sync(self):
         if not self.dependencies:
             return True
 
-        from hatchling.dep.core import dependencies_in_sync
+        from hatch.dep.sync import dependencies_in_sync
 
         return dependencies_in_sync(
             self.dependencies_complex, sys_path=self.python_info.sys_path, environment=self.python_info.environment

src/hatch/env/virtual.py

@@ -184,7 +184,7 @@ def dependencies_in_sync(self):
         if not self.dependencies:
             return True
 
-        from hatchling.dep.core import dependencies_in_sync
+        from hatch.dep.sync import dependencies_in_sync
 
         with self.safe_activation():
             return dependencies_in_sync(
@@ -199,7 +199,7 @@ def sync_dependencies(self):
     def build_environment(self, dependencies):
         from packaging.requirements import Requirement
 
-        from hatchling.dep.core import dependencies_in_sync
+        from hatch.dep.sync import dependencies_in_sync
 
         if not self.build_environment_exists():
             with self.expose_uv():

src/hatch/utils/dep.py

@@ -37,7 +37,7 @@ def hash_dependencies(requirements: list[Requirement]) -> str:
 def get_project_dependencies_complex(
     environment: EnvironmentInterface,
 ) -> tuple[dict[str, Requirement], dict[str, dict[str, Requirement]]]:
-    from hatchling.dep.core import dependencies_in_sync
+    from hatch.dep.sync import dependencies_in_sync
 
     dependencies_complex = {}
     optional_dependencies_complex = {}

tests/backend/dep/test_core.py → tests/dep/test_sync.py

@@ -3,8 +3,8 @@
 import pytest
 from packaging.requirements import Requirement
 
+from hatch.dep.sync import dependencies_in_sync
 from hatch.venv.core import TempUVVirtualEnv, TempVirtualEnv
-from hatchling.dep.core import dependencies_in_sync
 
 
 def test_no_dependencies(platform):