chore(installation): remove type errors

pyproject.toml

@@ -116,11 +116,6 @@ enable_error_code = ["ignore-without-code"]
 
 [[tool.mypy.overrides]]
 module = [
-  'poetry.installation.chef',
-  'poetry.installation.chooser',
-  'poetry.installation.executor',
-  'poetry.installation.installer',
-  'poetry.installation.pip_installer',
   'poetry.utils.env',
 ]
 ignore_errors = true
@@ -133,6 +128,7 @@ ignore_errors = true
 # warning.
 [[tool.mypy.overrides]]
 module = [
+  'poetry.installation.executor',
   'poetry.repositories.installed_repository',
 ]
 warn_unused_ignores = false

src/poetry/installation/chef.py

@@ -51,7 +51,7 @@ def get_cached_archive_for_link(self, link: Link) -> Link | None:
         if not archives:
             return link
 
-        candidates = []
+        candidates: list[tuple[float | None, Link]] = []
         for archive in archives:
             if not archive.is_wheel:
                 candidates.append((float("inf"), archive))

src/poetry/installation/chooser.py

@@ -4,6 +4,7 @@
 import re
 
 from typing import TYPE_CHECKING
+from typing import Any
 
 from packaging.tags import Tag
 
@@ -15,6 +16,7 @@
 if TYPE_CHECKING:
     from poetry.core.packages.package import Package
     from poetry.core.packages.utils.link import Link
+    from poetry.core.semver.version import Version
 
     from poetry.repositories.pool import Pool
     from poetry.utils.env import Env
@@ -144,7 +146,9 @@ def _get_links(self, package: Package) -> list[Link]:
 
         return selected_links
 
-    def _sort_key(self, package: Package, link: Link) -> tuple:
+    def _sort_key(
+        self, package: Package, link: Link
+    ) -> tuple[int, int, int, Version, tuple[Any, ...], int]:
         """
         Function to pass as the `key` argument to a call to sorted() to sort
         InstallationCandidates by preference.
@@ -168,7 +172,7 @@ def _sort_key(self, package: Package, link: Link) -> tuple:
               comparison operators, but then different sdist links
               with the same version, would have to be considered equal
         """
-        build_tag = ()
+        build_tag: tuple[Any, ...] = ()
         binary_preference = 0
         if link.is_wheel:
             wheel = Wheel(link.filename)
@@ -179,9 +183,11 @@ def _sort_key(self, package: Package, link: Link) -> tuple:
                 )
 
             # TODO: Binary preference
-            pri = -(wheel.get_minimum_supported_index(self._env.supported_tags))
+            pri = -(wheel.get_minimum_supported_index(self._env.supported_tags) or 0)
             if wheel.build_tag is not None:
                 match = re.match(r"^(\d+)(.*)$", wheel.build_tag)
+                if not match:
+                    raise ValueError(f"Unable to parse build tag: {wheel.build_tag}")
                 build_tag_groups = match.groups()
                 build_tag = (int(build_tag_groups[0]), build_tag_groups[1])
         else:  # sdist

src/poetry/installation/executor.py

@@ -12,6 +12,7 @@
 from subprocess import CalledProcessError
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import cast
 
 from cleo.io.null_io import NullIO
 from poetry.core.packages.file_dependency import FileDependency
@@ -21,6 +22,9 @@
 
 from poetry.installation.chef import Chef
 from poetry.installation.chooser import Chooser
+from poetry.installation.operations import Install
+from poetry.installation.operations import Uninstall
+from poetry.installation.operations import Update
 from poetry.utils._compat import decode
 from poetry.utils.authenticator import Authenticator
 from poetry.utils.env import EnvCommandError
@@ -31,12 +35,10 @@
 
 if TYPE_CHECKING:
     from cleo.io.io import IO
+    from cleo.io.outputs.section_output import SectionOutput
     from poetry.core.packages.package import Package
 
     from poetry.config.config import Config
-    from poetry.installation.operations import Install
-    from poetry.installation.operations import Uninstall
-    from poetry.installation.operations import Update
     from poetry.installation.operations.operation import Operation
     from poetry.repositories import Pool
     from poetry.utils.env import Env
@@ -49,7 +51,7 @@ def __init__(
         pool: Pool,
         config: Config,
         io: IO,
-        parallel: bool = None,
+        parallel: bool | None = None,
     ) -> None:
         self._env = env
         self._io = io
@@ -75,7 +77,7 @@ def __init__(
         self._executed_operations = 0
         self._executed = {"install": 0, "update": 0, "uninstall": 0}
         self._skipped = {"install": 0, "update": 0, "uninstall": 0}
-        self._sections = {}
+        self._sections: dict[int, SectionOutput] = {}
         self._lock = threading.Lock()
         self._shutdown = False
         self._hashes: dict[str, str] = {}
@@ -186,7 +188,7 @@ def _get_max_workers(desired_max_workers: int | None = None) -> int:
         # (it raises a NotImplementedError), so, in this case, we assume
         # that the system only has one CPU.
         try:
-            default_max_workers = os.cpu_count() + 4
+            default_max_workers = (os.cpu_count() or 1) + 4
         except NotImplementedError:
             default_max_workers = 5
 
@@ -312,7 +314,7 @@ def _do_execute_operation(self, operation: Operation) -> int:
 
             return 0
 
-        result = getattr(self, f"_execute_{method}")(operation)
+        result: int = getattr(self, f"_execute_{method}")(operation)
 
         if result != 0:
             return result
@@ -373,21 +375,21 @@ def get_operation_message(
             source_operation_color += "_dark"
             package_color += "_dark"
 
-        if operation.job_type == "install":
+        if isinstance(operation, Install):
             return (
                 f"<{base_tag}>Installing"
                 f" <{package_color}>{operation.package.name}</{package_color}>"
                 f" (<{operation_color}>{operation.package.full_pretty_version}</>)</>"
             )
 
-        if operation.job_type == "uninstall":
+        if isinstance(operation, Uninstall):
             return (
                 f"<{base_tag}>Removing"
                 f" <{package_color}>{operation.package.name}</{package_color}>"
                 f" (<{operation_color}>{operation.package.full_pretty_version}</>)</>"
             )
 
-        if operation.job_type == "update":
+        if isinstance(operation, Update):
             return (
                 f"<{base_tag}>Updating"
                 f" <{package_color}>{operation.initial_package.name}</{package_color}> "
@@ -643,7 +645,7 @@ def _validate_archive_hash(archive: Path | Link, package: Package) -> str:
             package.name,
             archive_path,
         )
-        archive_hash = "sha256:" + file_dep.hash()
+        archive_hash: str = "sha256:" + file_dep.hash()
         known_hashes = {f["hash"] for f in package.files}
 
         if archive_hash not in known_hashes:
@@ -681,7 +683,7 @@ def _download_archive(self, operation: Install | Update, link: Link) -> Path:
                 progress.start()
 
         done = 0
-        archive = self._chef.get_cache_directory_for_link(link) / link.filename
+        archive: Path = self._chef.get_cache_directory_for_link(link) / link.filename
         archive.parent.mkdir(parents=True, exist_ok=True)
         with archive.open("wb") as f:
             for chunk in response.iter_content(chunk_size=4096):
@@ -730,7 +732,7 @@ def _save_url_reference(self, operation: Operation) -> None:
                     direct_url_json.unlink()
             return
 
-        url_reference = None
+        url_reference: dict[str, Any] | None = None
 
         if package.source_type == "git":
             url_reference = self._create_git_url_reference(package)
@@ -745,26 +747,16 @@ def _save_url_reference(self, operation: Operation) -> None:
             for dist in self._env.site_packages.distributions(
                 name=package.name, writable_only=True
             ):
-                dist._path.joinpath("direct_url.json").write_text(
-                    json.dumps(url_reference),
-                    encoding="utf-8",
-                )
+                dist_path = cast(Path, dist._path)  # type: ignore[attr-defined]
+                url = dist_path / "direct_url.json"
+                url.write_text(json.dumps(url_reference), encoding="utf-8")
 
-                record = dist._path.joinpath("RECORD")
+                record = dist_path / "RECORD"
                 if record.exists():
                     with record.open(mode="a", encoding="utf-8") as f:
                         writer = csv.writer(f)
-                        writer.writerow(
-                            [
-                                str(
-                                    dist._path.joinpath("direct_url.json").relative_to(
-                                        record.parent.parent
-                                    )
-                                ),
-                                "",
-                                "",
-                            ]
-                        )
+                        path = url.relative_to(record.parent.parent)
+                        writer.writerow([str(path), "", ""])
 
     def _create_git_url_reference(
         self, package: Package
@@ -800,24 +792,20 @@ def _create_file_url_reference(
         if package.name in self._hashes:
             archive_info["hash"] = self._hashes[package.name]
 
-        reference = {
+        return {
             "url": Path(package.source_url).as_uri(),
             "archive_info": archive_info,
         }
 
-        return reference
-
     def _create_directory_url_reference(
         self, package: Package
-    ) -> dict[str, str | dict[str, str]]:
+    ) -> dict[str, str | dict[str, bool]]:
         dir_info = {}
 
         if package.develop:
             dir_info["editable"] = True
 
-        reference = {
+        return {
             "url": Path(package.source_url).as_uri(),
             "dir_info": dir_info,
         }
-
-        return reference

src/poetry/installation/installer.py

@@ -58,9 +58,9 @@ def __init__(
         self._execute_operations = True
         self._lock = False
 
-        self._whitelist = []
+        self._whitelist: list[str] = []
 
-        self._extras = []
+        self._extras: list[str] = []
 
         if executor is None:
             executor = Executor(self._env, self._pool, config, self._io)
@@ -171,7 +171,7 @@ def whitelist(self, packages: Iterable[str]) -> Installer:
 
         return self
 
-    def extras(self, extras: list) -> Installer:
+    def extras(self, extras: list[str]) -> Installer:
         self._extras = extras
 
         return self
@@ -182,7 +182,7 @@ def use_executor(self, use_executor: bool = True) -> Installer:
         return self
 
     def _do_refresh(self) -> int:
-        from poetry.puzzle import Solver
+        from poetry.puzzle.solver import Solver
 
         # Checking extras
         for extra in self._extras:
@@ -211,7 +211,7 @@ def _do_refresh(self) -> int:
         return 0
 
     def _do_install(self, local_repo: Repository) -> int:
-        from poetry.puzzle import Solver
+        from poetry.puzzle.solver import Solver
 
         locked_repository = Repository()
         if self._update:
@@ -475,9 +475,9 @@ def _populate_local_repo(
 
     def _get_operations_from_lock(
         self, locked_repository: Repository
-    ) -> Sequence[Operation]:
+    ) -> list[Operation]:
         installed_repo = self._installed_repository
-        ops = []
+        ops: list[Operation] = []
 
         extra_packages = self._get_extra_packages(locked_repository)
         for locked in locked_repository.packages:

src/poetry/installation/pip_installer.py

@@ -84,6 +84,7 @@ def install(self, package: Package, update: bool = False) -> None:
         if update:
             args.append("-U")
 
+        req: str | list[str]
         if package.files and not package.source_url:
             # Format as a requirements.txt
             # We need to create a requirements.txt file
@@ -136,10 +137,10 @@ def remove(self, package: Package) -> None:
             if src_dir.exists():
                 remove_directory(src_dir, force=True)
 
-    def run(self, *args: Any, **kwargs: Any) -> str:
+    def run(self, *args: Any, **kwargs: Any) -> int | str:
         return self._env.run_pip(*args, **kwargs)
 
-    def requirement(self, package: Package, formatted: bool = False) -> str:
+    def requirement(self, package: Package, formatted: bool = False) -> str | list[str]:
         if formatted and not package.source_type:
             req = f"{package.name}=={package.version}"
             for f in package.files:
@@ -161,7 +162,7 @@ def requirement(self, package: Package, formatted: bool = False) -> str:
                 req = os.path.realpath(package.source_url)
 
             if package.develop and package.source_type == "directory":
-                req = ["-e", req]
+                return ["-e", req]
 
             return req
 
@@ -172,7 +173,7 @@ def requirement(self, package: Package, formatted: bool = False) -> str:
             )
 
             if package.develop:
-                req = ["-e", req]
+                return ["-e", req]
 
             return req
 
@@ -183,9 +184,12 @@ def requirement(self, package: Package, formatted: bool = False) -> str:
 
     def create_temporary_requirement(self, package: Package) -> str:
         fd, name = tempfile.mkstemp("reqs.txt", f"{package.name}-{package.version}")
+        req = self.requirement(package, formatted=True)
+        if isinstance(req, list):
+            req = " ".join(req)
 
         try:
-            os.write(fd, encode(self.requirement(package, formatted=True)))
+            os.write(fd, encode(req))
         finally:
             os.close(fd)
 
@@ -237,7 +241,7 @@ def install_directory(self, package: Package) -> str | int:
                 with builder.setup_py():
                     if package.develop:
                         return pip_install(
-                            directory=req,
+                            path=req,
                             environment=self._env,
                             upgrade=True,
                             editable=True,
@@ -248,7 +252,7 @@ def install_directory(self, package: Package) -> str | int:
 
         if package.develop:
             return pip_install(
-                directory=req, environment=self._env, upgrade=True, editable=True
+                path=req, environment=self._env, upgrade=True, editable=True
             )
         return pip_install(path=req, environment=self._env, deps=False, upgrade=True)