Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,15 @@ repos:
hooks:
- id: ruff

- repo: https://github.com/pre-commit/mirrors-mypy
rev: v2.4.0
hooks:
- id: mypy
# Type-check pipenv's own code only; vendor/ and patched/ are excluded
# in [tool.mypy] and imports into them are skipped (follow_imports).
pass_filenames: false
args: []

- repo: https://github.com/pre-commit/pygrep-hooks
rev: v1.10.0
hooks:
Expand Down
5 changes: 5 additions & 0 deletions docs/dev/contributing.md
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,11 @@ and install the pre-commit hooks locally:
$ pre-commit run --all-files --verbose
```

The pre-commit hooks include [mypy](https://mypy.readthedocs.io/), which type-checks
pipenv's own code (`pipenv/vendor/` and `pipenv/patched/` are excluded). Its
settings live under `[tool.mypy]` in `pyproject.toml`, so you can also run it
directly with `mypy`.

### Code Review

Contributions will not be merged until they have been code reviewed. You should
Expand Down
1 change: 1 addition & 0 deletions news/5335.process.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Enable mypy type checking for pipenv's own code (excluding ``pipenv/vendor`` and ``pipenv/patched``) as a pre-commit hook, which also gates CI. Existing type errors were fixed, mostly by correcting annotations and converting legacy ``# type:`` comments to inline annotations.
3 changes: 2 additions & 1 deletion pipenv/cli/options.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import os
import re
from pathlib import Path
from typing import NoReturn

from pipenv.project import Project
from pipenv.utils.internet import is_valid_url
Expand Down Expand Up @@ -500,7 +501,7 @@ class _PipenvArgumentParser(argparse.ArgumentParser):
"""ArgumentParser subclass that appends 'Did you mean …?' suggestions
when the user types an unrecognised subcommand name."""

def error(self, message: str) -> None:
def error(self, message: str) -> NoReturn:
import difflib

m = re.search(r"invalid choice: '([^']+)'", message)
Expand Down
101 changes: 53 additions & 48 deletions pipenv/environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@

if typing.TYPE_CHECKING:
from types import ModuleType
from typing import ContextManager, Generator
from typing import Any, Generator, Iterator

from pipenv.project import Project, TPipfile, TSource
from pipenv.vendor import tomlkit
Expand All @@ -54,18 +54,18 @@
class Environment:
def __init__(
self,
prefix: str | None = None,
prefix: str | Path | None = None,
python: str | None = None,
is_venv: bool = False,
base_working_set: list[importlib_metadata.Distribution] = None,
base_working_set: Iterable[importlib_metadata.Distribution] | None = None,
pipfile: tomlkit.toml_document.TOMLDocument | TPipfile | None = None,
sources: list[TSource] | None = None,
project: Project | None = None,
):
super().__init__()
self._modules = {"pipenv": pipenv}
self.base_working_set = base_working_set if base_working_set else BASE_WORKING_SET
prefix = normalize_path(prefix)
prefix = normalize_path(prefix if prefix else sys.prefix)
self._python = None
if python is not None:
self._python = Path(python).absolute().as_posix()
Expand All @@ -79,30 +79,20 @@ def __init__(
if project and not pipfile:
pipfile = project.pipfile.parsed
self.pipfile = pipfile
self.extra_dists = []
self.extra_dists: list[importlib_metadata.Distribution] = []
if self.is_venv and prefix is not None and not Path(prefix).exists():
return
self.prefix = Path(prefix if prefix else sys.prefix)
self._base_paths = {}
self._base_paths: dict[str, Any] | None = {}
if self.is_venv:
self._base_paths = self.get_paths()
self.sys_paths = get_paths()

def safe_import(self, name: str) -> ModuleType:
"""Helper utility for reimporting previously imported modules while inside the env"""
module = None
if name not in self._modules:
self._modules[name] = importlib.import_module(name)
module = self._modules[name]
if not module:
dist = next(
iter(dist for dist in self.base_working_set if dist.project_name == name),
None,
)
if dist:
dist.activate()
module = importlib.import_module(name)
return module
return self._modules[name]

@cached_property
def python_version(self) -> str | None:
Expand Down Expand Up @@ -157,7 +147,7 @@ def install_scheme(self):
return "posix_prefix"

@cached_property
def base_paths(self) -> dict[str, str]:
def base_paths(self) -> dict[str, Any]:
"""
Returns the context appropriate paths for the environment.

Expand All @@ -182,19 +172,21 @@ def base_paths(self) -> dict[str, str]:
"""

prefix = Path(self.prefix)
paths = {}
paths: dict[str, Any] = {}
if self._base_paths:
paths = self._base_paths.copy()
else:
try:
paths = self.get_paths()
paths = self.get_paths() or {}
except Exception:
paths = get_paths(
self.install_scheme,
vars={
"base": prefix,
"platbase": prefix,
},
paths.update(
get_paths(
self.install_scheme,
vars={
"base": prefix,
"platbase": prefix,
},
)
)
current_version = get_python_version()
try:
Expand All @@ -207,16 +199,18 @@ def base_paths(self) -> dict[str, str]:
# Sometimes virtualenvs are made using virtualenv interpreters and there is no
# include directory, which will cause this approach to fail. This failsafe
# will make sure we fall back to the shell execution to find the real include path
paths = self.get_include_path()
paths = self.get_include_path() or {}
paths.update(self.get_lib_paths())
paths["scripts"] = self.script_basedir
if not paths:
paths = get_paths(
self.install_scheme,
vars={
"base": prefix,
"platbase": prefix,
},
paths.update(
get_paths(
self.install_scheme,
vars={
"base": prefix,
"platbase": prefix,
},
)
)
if not os.path.exists(paths["purelib"]) and not os.path.exists(paths["platlib"]):
lib_paths = self.get_lib_paths()
Expand Down Expand Up @@ -329,7 +323,7 @@ def build_command(
py_command = py_command % lines_as_str
return py_command

def get_paths(self) -> dict[str, str] | None:
def get_paths(self) -> dict[str, Any] | None:
"""
Get the paths for the environment by running a subcommand

Expand Down Expand Up @@ -412,7 +406,7 @@ def get_lib_paths(self) -> dict[str, str]:
return paths
return {}

def get_include_path(self) -> dict[str, str] | None:
def get_include_path(self) -> dict[str, Any] | None:
"""Get the include path for the environment

:return: The python include path for the environment
Expand Down Expand Up @@ -464,7 +458,7 @@ def scripts_dir(self) -> str:
return self.paths["scripts"]

@property
def libdir(self) -> str:
def libdir(self) -> tuple[str, str]:
purelib = self.paths.get("purelib", None)
if purelib and os.path.exists(purelib):
return "purelib", purelib
Expand Down Expand Up @@ -507,12 +501,13 @@ def get_distributions(self) -> Generator[importlib_metadata.Distribution, None,
def find_egg(self, egg_dist: importlib_metadata.Distribution) -> str | None:
"""Find an egg by name in the given environment"""
site_packages = self.libdir[1]
search_filename = f"{egg_dist._normalized_name}.egg-link"
search_filename = f"{egg_dist._normalized_name}.egg-link" # type: ignore[attr-defined]
user_site: str | None
try:
user_site = site.getusersitepackages()
except AttributeError:
user_site = site.USER_SITE
search_locations = [site_packages, user_site]
search_locations = [loc for loc in (site_packages, user_site) if loc]
for site_directory in search_locations:
egg = os.path.join(site_directory, search_filename)
if os.path.isfile(egg):
Expand All @@ -525,7 +520,7 @@ def locate_dist(self, dist: importlib_metadata.Distribution) -> str:
If the egg - link doesn 't exist, return the supplied distribution."""

location = self.find_egg(dist)
return location or dist._path
return location or dist._path # type: ignore[attr-defined]

def dist_is_in_project(self, dist: importlib_metadata.Distribution) -> bool:
"""Determine whether the supplied distribution is in the environment."""
Expand All @@ -548,19 +543,21 @@ def get_installed_packages(self) -> list[importlib_metadata.Distribution]:
packages = [
pkg
for pkg in workingset
if self.dist_is_in_project(pkg) and pkg._normalized_name != "python"
if self.dist_is_in_project(pkg)
and pkg._normalized_name != "python" # type: ignore[attr-defined]
]
return packages

@contextlib.contextmanager
def get_finder(self, pre: bool = False) -> ContextManager[PackageFinder]:
def get_finder(self, pre: bool = False) -> Iterator[PackageFinder]:
from pipenv.patched.pip._internal.commands.install import InstallCommand

from .utils.resolver import get_package_finder

pip_command = InstallCommand(
name="InstallCommand", summary="pip Install command."
)
assert self.project is not None and self.pipfile is not None
pip_args = prepare_pip_source_args(self.sources)
pip_options, _ = pip_command.parser.parse_args(pip_args)
pip_options.cache_dir = self.project.s.PIPENV_CACHE_DIR
Expand All @@ -581,11 +578,12 @@ def get_finder(self, pre: bool = False) -> ContextManager[PackageFinder]:
def get_package_info(
self, pre: bool = False
) -> Generator[importlib_metadata.Distribution, None, None]:
assert self.pipfile is not None
packages = self.get_installed_packages()

with self.get_finder() as finder:
for dist in packages:
name = dist._normalized_name
name = dist._normalized_name # type: ignore[attr-defined]
all_candidates = finder.find_all_candidates(name)
allow_prereleases = self.pipfile.get("pre", False)
if not allow_prereleases and finder.release_control is not None:
Expand Down Expand Up @@ -614,7 +612,7 @@ def get_package_info(
else:
pass
# This is dirty but makes the rest of the code much cleaner
dist.latest_version = remote_version
dist.latest_version = remote_version # type: ignore[attr-defined]
yield dist

def get_outdated_packages(
Expand All @@ -623,7 +621,7 @@ def get_outdated_packages(
return [
pkg
for pkg in self.get_package_info(pre=pre)
if pkg.latest_version > parse_version(pkg.version)
if pkg.latest_version > parse_version(pkg.version) # type: ignore[attr-defined]
]

@classmethod
Expand Down Expand Up @@ -735,15 +733,20 @@ def is_installed(self, pkgname):
:rtype: bool
"""

return any(d for d in self.get_distributions() if d._normalized_name == pkgname)
return any(
d
for d in self.get_distributions()
if d._normalized_name == pkgname # type: ignore[attr-defined]
)

def is_satisfied(self, req: InstallRequirement):
match = next(
iter(
d
for d in self.get_distributions()
if req.name
and canonicalize_name(d._normalized_name) == canonicalize_name(req.name)
and canonicalize_name(d._normalized_name) # type: ignore[attr-defined]
== canonicalize_name(req.name)
),
None,
)
Expand All @@ -767,8 +770,10 @@ def is_satisfied(self, req: InstallRequirement):
else:
parsed_url = urlparse(requested_path)
local_path = parsed_url.path
return requested_path and os.path.samefile(local_path, match.location)
elif match.has_metadata("direct_url.json"):
return requested_path and os.path.samefile(
local_path, str(match.locate_file(""))
)
elif match.read_text("direct_url.json") is not None:
# Direct URL installs we assume are not satisfied since we may be
# installing from Pipfile and have insufficient information to determine
# if the content has changed.
Expand Down
12 changes: 6 additions & 6 deletions pipenv/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import itertools
import sys
from collections import namedtuple
from traceback import format_tb
from typing import NamedTuple, Optional

from pipenv.patched.pip._vendor.rich.console import Console
from pipenv.patched.pip._vendor.rich.text import Text
Expand Down Expand Up @@ -60,11 +60,11 @@ def unstyle(text: str) -> str:
return str(text)


KnownException = namedtuple(
"KnownException",
["exception_name", "match_string", "show_from_string", "prefix"],
)
KnownException.__new__.__defaults__ = (None, None, None, "")
class KnownException(NamedTuple):
exception_name: Optional[str] = None
match_string: Optional[str] = None
show_from_string: Optional[str] = None
prefix: str = ""

KNOWN_EXCEPTIONS = [
KnownException("PermissionError", prefix="Permission Denied:"),
Expand Down
1 change: 1 addition & 0 deletions pipenv/project.py
Original file line number Diff line number Diff line change
Expand Up @@ -267,6 +267,7 @@ def installed_package_names(self):

def get_environment(self, allow_global: bool = False) -> Environment:
is_venv = is_in_virtualenv()
prefix: str | Path
if allow_global and not is_venv:
prefix = sys.prefix
python = sys.executable
Expand Down
1 change: 1 addition & 0 deletions pipenv/resolver/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ def _ensure_modules() -> None:
"typing_extensions",
location=typing_ext_path,
)
assert spec is not None and spec.loader is not None
typing_extensions = importlib.util.module_from_spec(spec)
sys.modules["typing_extensions"] = typing_extensions
spec.loader.exec_module(typing_extensions)
Expand Down
3 changes: 1 addition & 2 deletions pipenv/resolver/pure_python_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -478,6 +478,7 @@ def _fetch_via_wheel_head(session: Any, wheel_url: str) -> CoreMetadata:
)
try:
zf = zipfile.ZipFile(partial)
break
except zipfile.BadZipFile:
if window >= total_length or window >= _MAX_CENTRAL_DIRECTORY_WINDOW:
raise MetadataFetchError(
Expand All @@ -486,8 +487,6 @@ def _fetch_via_wheel_head(session: Any, wheel_url: str) -> CoreMetadata:
)
# Double the window and retry.
window = min(window * 2, total_length, _MAX_CENTRAL_DIRECTORY_WINDOW)
continue
break

try:
info = _find_metadata_member(zf)
Expand Down
2 changes: 1 addition & 1 deletion pipenv/resolver/pure_python_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -1131,7 +1131,7 @@ def is_satisfied_by(self, requirement, candidate) -> bool: ...
# ``Provides-Extra``. Empty frozenset → metadata not loaded;
# admit per the lazy-metadata clause above. Non-empty →
# strict subset check.
candidate_extras = getattr(candidate, "extras", frozenset())
candidate_extras: frozenset[str] = getattr(candidate, "extras", frozenset())
if candidate_extras and not requirement.extras <= candidate_extras:
return False

Expand Down
2 changes: 1 addition & 1 deletion pipenv/resolver/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -292,7 +292,7 @@ def from_install_requirement(
entry: dict = {"name": name}

# ---- VCS branch (locking.py:60-83) ----
is_vcs_dep = next(
is_vcs_dep: str | bool | None = next(
iter([v for v in VCS_LIST if v in pipfile_entry]), None
)
if getattr(req, "link", None) is not None and getattr(req.link, "is_vcs", False):
Expand Down
Loading
Loading