refactor: split off functions to get direct origin dependencies from provider (#7693)

src/poetry/packages/direct_origin.py

+from __future__ import annotations
+
+import functools
+import os
+import tempfile
+import urllib.parse
+
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from poetry.inspection.info import PackageInfo
+from poetry.inspection.info import PackageInfoError
+from poetry.utils.helpers import download_file
+from poetry.utils.helpers import get_file_hash
+from poetry.vcs.git import Git
+
+
+if TYPE_CHECKING:
+    from poetry.core.packages.package import Package
+
+
+@functools.lru_cache(maxsize=None)
+def _get_package_from_git(
+    url: str,
+    branch: str | None = None,
+    tag: str | None = None,
+    rev: str | None = None,
+    subdirectory: str | None = None,
+    source_root: Path | None = None,
+) -> Package:
+    source = Git.clone(
+        url=url,
+        source_root=source_root,
+        branch=branch,
+        tag=tag,
+        revision=rev,
+        clean=False,
+    )
+    revision = Git.get_revision(source)
+
+    path = Path(source.path)
+    if subdirectory:
+        path = path.joinpath(subdirectory)
+
+    package = DirectOrigin.get_package_from_directory(path)
+    package._source_type = "git"
+    package._source_url = url
+    package._source_reference = rev or tag or branch or "HEAD"
+    package._source_resolved_reference = revision
+    package._source_subdirectory = subdirectory
+
+    return package
+
+
+class DirectOrigin:
+    @classmethod
+    def get_package_from_file(cls, file_path: Path) -> Package:
+        try:
+            package = PackageInfo.from_path(path=file_path).to_package(
+                root_dir=file_path
+            )
+        except PackageInfoError:
+            raise RuntimeError(
+                f"Unable to determine package info from path: {file_path}"
+            )
+
+        return package
+
+    @classmethod
+    def get_package_from_directory(cls, directory: Path) -> Package:
+        return PackageInfo.from_directory(path=directory).to_package(root_dir=directory)
+
+    @classmethod
+    def get_package_from_url(cls, url: str) -> Package:
+        file_name = os.path.basename(urllib.parse.urlparse(url).path)
+        with tempfile.TemporaryDirectory() as temp_dir:
+            dest = Path(temp_dir) / file_name
+            download_file(url, dest)
+            package = cls.get_package_from_file(dest)
+
+            package.files = [
+                {"file": file_name, "hash": "sha256:" + get_file_hash(dest)}
+            ]
+
+        package._source_type = "url"
+        package._source_url = url
+
+        return package
+
+    @staticmethod
+    def get_package_from_vcs(
+        vcs: str,
+        url: str,
+        branch: str | None = None,
+        tag: str | None = None,
+        rev: str | None = None,
+        subdirectory: str | None = None,
+        source_root: Path | None = None,
+    ) -> Package:
+        if vcs != "git":
+            raise ValueError(f"Unsupported VCS dependency {vcs}")
+
+        return _get_package_from_git(
+            url=url,
+            branch=branch,
+            tag=tag,
+            rev=rev,
+            subdirectory=subdirectory,
+            source_root=source_root,
+        )

src/poetry/puzzle/provider.py

 from __future__ import annotations
 
-import functools
 import logging
-import os
 import re
-import tempfile
 import time
-import urllib.parse
 
 from collections import defaultdict
 from contextlib import contextmanager
-from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import cast
 
@@ -22,19 +17,16 @@ from poetry.core.version.markers import AnyMarker
 from poetry.core.version.markers import EmptyMarker
 from poetry.core.version.markers import MarkerUnion
 
-from poetry.inspection.info import PackageInfo
-from poetry.inspection.info import PackageInfoError
 from poetry.mixology.incompatibility import Incompatibility
 from poetry.mixology.incompatibility_cause import DependencyCause
 from poetry.mixology.incompatibility_cause import PythonCause
 from poetry.mixology.term import Term
 from poetry.packages import DependencyPackage
+from poetry.packages.direct_origin import DirectOrigin
 from poetry.packages.package_collection import PackageCollection
 from poetry.puzzle.exceptions import OverrideNeeded
 from poetry.repositories.exceptions import PackageNotFound
-from poetry.utils.helpers import download_file
 from poetry.utils.helpers import get_file_hash
-from poetry.vcs.git import Git
 
 
 if TYPE_CHECKING:
@@ -42,6 +34,7 @@ if TYPE_CHECKING:
     from collections.abc import Collection
     from collections.abc import Iterable
     from collections.abc import Iterator
+    from pathlib import Path
 
     from cleo.io.io import IO
     from packaging.utils import NormalizedName
@@ -100,39 +93,6 @@ class Indicator(ProgressIndicator):
         return f"{elapsed:.1f}s"
 
 
-@functools.lru_cache(maxsize=None)
-def _get_package_from_git(
-    url: str,
-    branch: str | None = None,
-    tag: str | None = None,
-    rev: str | None = None,
-    subdirectory: str | None = None,
-    source_root: Path | None = None,
-) -> Package:
-    source = Git.clone(
-        url=url,
-        source_root=source_root,
-        branch=branch,
-        tag=tag,
-        revision=rev,
-        clean=False,
-    )
-    revision = Git.get_revision(source)
-
-    path = Path(source.path)
-    if subdirectory:
-        path = path.joinpath(subdirectory)
-
-    package = Provider.get_package_from_directory(path)
-    package._source_type = "git"
-    package._source_url = url
-    package._source_reference = rev or tag or branch or "HEAD"
-    package._source_resolved_reference = revision
-    package._source_subdirectory = subdirectory
-
-    return package
-
-
 class Provider:
     UNSAFE_PACKAGES: set[str] = set()
 
@@ -351,7 +311,7 @@ class Provider:
         Basically, we clone the repository in a temporary directory
         and get the information we need by checking out the specified reference.
         """
-        package = self.get_package_from_vcs(
+        package = DirectOrigin.get_package_from_vcs(
             dependency.vcs,
             dependency.source,
             branch=dependency.branch,
@@ -368,31 +328,9 @@ class Provider:
 
         return package
 
-    @staticmethod
-    def get_package_from_vcs(
-        vcs: str,
-        url: str,
-        branch: str | None = None,
-        tag: str | None = None,
-        rev: str | None = None,
-        subdirectory: str | None = None,
-        source_root: Path | None = None,
-    ) -> Package:
-        if vcs != "git":
-            raise ValueError(f"Unsupported VCS dependency {vcs}")
-
-        return _get_package_from_git(
-            url=url,
-            branch=branch,
-            tag=tag,
-            rev=rev,
-            subdirectory=subdirectory,
-            source_root=source_root,
-        )
-
     def _search_for_file(self, dependency: FileDependency) -> Package:
         dependency.validate(raise_error=True)
-        package = self.get_package_from_file(dependency.full_path)
+        package = DirectOrigin.get_package_from_file(dependency.full_path)
 
         self.validate_package_for_dependency(dependency=dependency, package=package)
 
@@ -408,22 +346,9 @@ class Provider:
 
         return package
 
-    @classmethod
-    def get_package_from_file(cls, file_path: Path) -> Package:
-        try:
-            package = PackageInfo.from_path(path=file_path).to_package(
-                root_dir=file_path
-            )
-        except PackageInfoError:
-            raise RuntimeError(
-                f"Unable to determine package info from path: {file_path}"
-            )
-
-        return package
-
     def _search_for_directory(self, dependency: DirectoryDependency) -> Package:
         dependency.validate(raise_error=True)
-        package = self.get_package_from_directory(dependency.full_path)
+        package = DirectOrigin.get_package_from_directory(dependency.full_path)
 
         self.validate_package_for_dependency(dependency=dependency, package=package)
 
@@ -434,12 +359,8 @@ class Provider:
 
         return package
 
-    @classmethod
-    def get_package_from_directory(cls, directory: Path) -> Package:
-        return PackageInfo.from_directory(path=directory).to_package(root_dir=directory)
-
     def _search_for_url(self, dependency: URLDependency) -> Package:
-        package = self.get_package_from_url(dependency.url)
+        package = DirectOrigin.get_package_from_url(dependency.url)
 
         self.validate_package_for_dependency(dependency=dependency, package=package)
 
@@ -453,23 +374,6 @@ class Provider:
 
         return package
 
-    @classmethod
-    def get_package_from_url(cls, url: str) -> Package:
-        file_name = os.path.basename(urllib.parse.urlparse(url).path)
-        with tempfile.TemporaryDirectory() as temp_dir:
-            dest = Path(temp_dir) / file_name
-            download_file(url, dest)
-            package = cls.get_package_from_file(dest)
-
-            package.files = [
-                {"file": file_name, "hash": "sha256:" + get_file_hash(dest)}
-            ]
-
-        package._source_type = "url"
-        package._source_url = url
-
-        return package
-
     def _get_dependencies_with_overrides(
         self, dependencies: list[Dependency], package: DependencyPackage
     ) -> list[Dependency]:

src/poetry/utils/dependency_specification.py

@@ -16,7 +16,7 @@ from typing import cast
 from poetry.core.packages.dependency import Dependency
 from tomlkit.items import InlineTable
 
-from poetry.puzzle.provider import Provider
+from poetry.packages.direct_origin import DirectOrigin
 
 
 if TYPE_CHECKING:
@@ -120,7 +120,7 @@ class RequirementsParser:
             pair["subdirectory"] = parsed.subdirectory
 
         source_root = self._env.path.joinpath("src") if self._env else None
-        package = Provider.get_package_from_vcs(
+        package = DirectOrigin.get_package_from_vcs(
             "git",
             url=url.url,
             rev=pair.get("rev"),
@@ -139,7 +139,7 @@ class RequirementsParser:
             return self._parse_git_url(requirement)
 
         if url_parsed.scheme in ["http", "https"]:
-            package = Provider.get_package_from_url(requirement)
+            package = DirectOrigin.get_package_from_url(requirement)
             assert package.source_url is not None
             return {"name": package.name, "url": package.source_url}
 
@@ -158,9 +158,9 @@ class RequirementsParser:
                 path = self._cwd.joinpath(requirement)
 
             if path.is_file():
-                package = Provider.get_package_from_file(path.resolve())
+                package = DirectOrigin.get_package_from_file(path.resolve())
             else:
-                package = Provider.get_package_from_directory(path.resolve())
+                package = DirectOrigin.get_package_from_directory(path.resolve())
 
             return {
                 "name": package.name,

tests/conftest.py

@@ -239,7 +239,7 @@ def mock_user_config_dir(mocker: MockerFixture, config_dir: Path) -> None:
 def download_mock(mocker: MockerFixture) -> None:
     # Patch download to not download anything but to just copy from fixtures
     mocker.patch("poetry.utils.helpers.download_file", new=mock_download)
-    mocker.patch("poetry.puzzle.provider.download_file", new=mock_download)
+    mocker.patch("poetry.packages.direct_origin.download_file", new=mock_download)
     mocker.patch("poetry.repositories.http_repository.download_file", new=mock_download)