Skip to content

P
python-poetry
Commit 2ae46fa5 by Branch Vincent Committed by Bjorn Neergaard
Options

chore(installation): remove type errors

parent e38c045c
Showing with 53 additions and 59 deletions
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 @@ class Chef:
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 logging
import re
from typing import TYPE_CHECKING
from typing import Any
from packaging.tags import Tag
......@@ -15,6 +16,7 @@ from poetry.utils.patterns import wheel_file_re
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 @@ class Chooser:
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 @@ class Chooser:
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 @@ class Chooser:
)
# 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 pathlib import Path
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.core.pyproject.toml import PyProjectTOML
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 @@ from poetry.utils.pip import pip_install
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 @@ class Executor:
pool: Pool,
config: Config,
io: IO,
parallel: bool = None,
parallel: bool | None = None,
) -> None:
self._env = env
self._io = io
......@@ -75,7 +77,7 @@ class Executor:
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 @@ class Executor:
# (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 @@ class Executor:
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 @@ class Executor:
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 @@ class Executor:
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 @@ class Executor:
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 @@ class Executor:
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 @@ class Executor:
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 @@ class Executor:
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 @@ class Installer:
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 @@ class 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 @@ class 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 @@ class Installer:
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 @@ class Installer:
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 @@ class PipInstaller(BaseInstaller):
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 @@ class PipInstaller(BaseInstaller):
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 @@ class PipInstaller(BaseInstaller):
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 @@ class PipInstaller(BaseInstaller):
)
if package.develop:
req = ["-e", req]
return ["-e", req]
return req
......@@ -183,9 +184,12 @@ class PipInstaller(BaseInstaller):
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 @@ class PipInstaller(BaseInstaller):
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 @@ class PipInstaller(BaseInstaller):
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)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment