diff --git a/pyproject.toml b/pyproject.toml index eb07d595b4d..d7a1db7b42e 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/poetry/installation/chef.py b/src/poetry/installation/chef.py index 0cf7b54e3b5..51cad799ab7 100644 --- a/src/poetry/installation/chef.py +++ b/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)) diff --git a/src/poetry/installation/chooser.py b/src/poetry/installation/chooser.py index 82659d444eb..ebfbbdaa765 100644 --- a/src/poetry/installation/chooser.py +++ b/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 diff --git a/src/poetry/installation/executor.py b/src/poetry/installation/executor.py index 8e1f18f78fa..3d64507a191 100644 --- a/src/poetry/installation/executor.py +++ b/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}" 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}" 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} " @@ -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 diff --git a/src/poetry/installation/installer.py b/src/poetry/installation/installer.py index 25b13092673..b033e71f53c 100644 --- a/src/poetry/installation/installer.py +++ b/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: diff --git a/src/poetry/installation/pip_installer.py b/src/poetry/installation/pip_installer.py index da01d507a16..ae54ff73af5 100644 --- a/src/poetry/installation/pip_installer.py +++ b/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)