McSinyx
/
pip
mirror of https://github.com/pypa/pip
1
1
Fork 0
Code Issues Releases Activity

Merge pull request #6093 from mkurnikov/req-types-finished 

Finish types for pip._internal.req, set disallow_untyped_defs flag
This commit is contained in:
Chris Jerdonek 2019-03-23 02:16:35 -07:00 committed by GitHub
parent d245434c02 8c70363262
commit 1bb21fd3ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 89 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
setup.cfg
View File

@ -27,6 +27,9 @@ follow_imports = silent
ignore_missing_imports = True
strict_optional = False
[mypy-pip/_internal/req/*]
disallow_untyped_defs = True
[mypy-pip/_vendor/*]
follow_imports = skip
ignore_errors = True

5
src/pip/_internal/req/__init__.py
View File

@ -9,7 +9,7 @@ from pip._internal.utils.logging import indent_log
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
if MYPY_CHECK_RUNNING:
from typing import List, Sequence
from typing import Any, List, Sequence
__all__ = [
"RequirementSet", "InstallRequirement",
@ -23,7 +23,8 @@ def install_given_reqs(
to_install, # type: List[InstallRequirement]
install_options, # type: List[str]
global_options=(), # type: Sequence[str]
*args, **kwargs
*args, # type: Any
**kwargs # type: Any
):
# type: (...) -> List[InstallRequirement]
"""

2
src/pip/_internal/req/constructors.py
View File

@ -32,7 +32,7 @@ from pip._internal.wheel import Wheel
if MYPY_CHECK_RUNNING:
from typing import (
Optional, Tuple, Set, Any, Union, Dict,
Any, Dict, Optional, Set, Tuple, Union,
)
from pip._internal.cache import WheelCache

3
src/pip/_internal/req/req_file.py
View File

@ -23,7 +23,7 @@ from pip._internal.utils.typing import MYPY_CHECK_RUNNING
if MYPY_CHECK_RUNNING:
from typing import (
Iterator, Tuple, Optional, List, Callable, Text
Any, Callable, Iterator, List, NoReturn, Optional, Text, Tuple,
)
from pip._internal.req import InstallRequirement
from pip._internal.cache import WheelCache
@ -288,6 +288,7 @@ def build_parser(line):
# By default optparse sys.exits on parsing errors. We want to wrap
# that in our own exception.
def parser_exit(self, msg):
# type: (Any, str) -> NoReturn
# add offending line
msg = 'Invalid requirement: %s\n%s' % (line, msg)
raise RequirementsFileParseError(msg)

15
src/pip/_internal/req/req_install.py
View File

@ -42,7 +42,7 @@ from pip._internal.wheel import move_wheel_files
if MYPY_CHECK_RUNNING:
from typing import (
Optional, Iterable, List, Union, Any, Sequence, Dict
Any, Dict, Iterable, List, Mapping, Optional, Sequence, Union,
)
from pip._internal.build_env import BuildEnvironment
from pip._internal.cache import WheelCache
@ -156,6 +156,7 @@ class InstallRequirement(object):
self.use_pep517 = use_pep517
def __str__(self):
# type: () -> str
if self.req:
s = str(self.req)
if self.link:
@ -176,6 +177,7 @@ class InstallRequirement(object):
return s
def __repr__(self):
# type: () -> str
return '<%s object: %s editable=%r>' % (
self.__class__.__name__, str(self), self.editable)
@ -226,6 +228,7 @@ class InstallRequirement(object):
@property
def installed_version(self):
# type: () -> Optional[str]
return get_installed_version(self.name)
def match_markers(self, extras_requested=None):
@ -501,7 +504,12 @@ class InstallRequirement(object):
# Use a custom function to call subprocesses
self.spin_message = ""
def runner(cmd, cwd=None, extra_environ=None):
def runner(
cmd, # type: List[str]
cwd=None, # type: Optional[str]
extra_environ=None # type: Optional[Mapping[str, Any]]
):
# type: (...) -> None
with open_spinner(self.spin_message) as spinner:
call_subprocess(
cmd,
@ -661,6 +669,7 @@ class InstallRequirement(object):
@property
def metadata(self):
# type: () -> Any
if not hasattr(self, '_metadata'):
self._metadata = get_metadata(self.get_dist())
@ -815,6 +824,7 @@ class InstallRequirement(object):
return uninstalled_pathset
def _clean_zip_name(self, name, prefix): # only used by archive.
# type: (str, str) -> str
assert name.startswith(prefix + os.path.sep), (
"name %r doesn't start with prefix %r" % (name, prefix)
)
@ -947,6 +957,7 @@ class InstallRequirement(object):
self.install_succeeded = True
def prepend_root(path):
# type: (str) -> str
if root is None or not os.path.isabs(path):
return path
else:

4
src/pip/_internal/req/req_set.py
View File

@ -9,7 +9,7 @@ from pip._internal.utils.typing import MYPY_CHECK_RUNNING
from pip._internal.wheel import Wheel
if MYPY_CHECK_RUNNING:
from typing import Optional, List, Tuple, Dict, Iterable
from typing import Dict, Iterable, List, Optional, Tuple
from pip._internal.req.req_install import InstallRequirement
@ -34,12 +34,14 @@ class RequirementSet(object):
self.reqs_to_cleanup = [] # type: List[InstallRequirement]
def __str__(self):
# type: () -> str
reqs = [req for req in self.requirements.values()
if not req.comes_from]
reqs.sort(key=lambda req: req.name.lower())
return ' '.join([str(req.req) for req in reqs])
def __repr__(self):
# type: () -> str
reqs = [req for req in self.requirements.values()]
reqs.sort(key=lambda req: req.name.lower())
reqs_str = ', '.join([str(req.req) for req in reqs])

12
src/pip/_internal/req/req_tracker.py
View File

@ -10,7 +10,8 @@ from pip._internal.utils.temp_dir import TempDirectory
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
if MYPY_CHECK_RUNNING:
from typing import Set, Iterator
from types import TracebackType
from typing import Iterator, Optional, Set, Type
from pip._internal.req.req_install import InstallRequirement
from pip._internal.models.link import Link
@ -33,9 +34,16 @@ class RequirementTracker(object):
self._entries = set() # type: Set[InstallRequirement]
def __enter__(self):
# type: () -> RequirementTracker
return self
def __exit__(self, exc_type, exc_val, exc_tb):
def __exit__(
self,
exc_type, # type: Optional[Type[BaseException]]
exc_val, # type: Optional[BaseException]
exc_tb # type: Optional[TracebackType]
):
# type: (...) -> None
self.cleanup()
def _entry_path(self, link):

71
src/pip/_internal/req/req_uninstall.py
View File

@ -18,11 +18,19 @@ from pip._internal.utils.misc import (
normalize_path, renames, rmtree,
)
from pip._internal.utils.temp_dir import AdjacentTempDirectory, TempDirectory
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
if MYPY_CHECK_RUNNING:
from typing import (
Any, Callable, Dict, Iterable, Iterator, List, Optional, Set, Tuple,
)
from pip._vendor.pkg_resources import Distribution
logger = logging.getLogger(__name__)
def _script_names(dist, script_name, is_gui):
# type: (Distribution, str, bool) -> List[str]
"""Create the fully qualified name of the files created by
{console,gui}_scripts for the given ``dist``.
Returns the list of file names
@ -44,9 +52,11 @@ def _script_names(dist, script_name, is_gui):
def _unique(fn):
# type: (Callable) -> Callable[..., Iterator[Any]]
@functools.wraps(fn)
def unique(*args, **kw):
seen = set()
# type: (Any, Any) -> Iterator[Any]
seen = set() # type: Set[Any]
for item in fn(*args, **kw):
if item not in seen:
seen.add(item)
@ -56,6 +66,7 @@ def _unique(fn):
@_unique
def uninstallation_paths(dist):
# type: (Distribution) -> Iterator[str]
"""
Yield all the uninstallation paths for dist based on RECORD-without-.py[co]
@ -78,13 +89,14 @@ def uninstallation_paths(dist):
def compact(paths):
# type: (Iterable[str]) -> Set[str]
"""Compact a path set to contain the minimal number of paths
necessary to contain all paths in the set. If /a/path/ and
/a/path/to/a/file.txt are both in the set, leave only the
shorter path."""
sep = os.path.sep
short_paths = set()
short_paths = set() # type: Set[str]
for path in sorted(paths, key=len):
should_skip = any(
path.startswith(shortpath.rstrip("*")) and
@ -97,6 +109,7 @@ def compact(paths):
def compress_for_rename(paths):
# type: (Iterable[str]) -> Set[str]
"""Returns a set containing the paths that need to be renamed.
This set may include directories when the original sequence of paths
@ -106,9 +119,10 @@ def compress_for_rename(paths):
remaining = set(case_map)
unchecked = sorted(set(os.path.split(p)[0]
for p in case_map.values()), key=len)
wildcards = set()
wildcards = set() # type: Set[str]
def norm_join(*a):
# type: (str) -> str
return os.path.normcase(os.path.join(*a))
for root in unchecked:
@ -117,8 +131,8 @@ def compress_for_rename(paths):
# This directory has already been handled.
continue
all_files = set()
all_subdirs = set()
all_files = set() # type: Set[str]
all_subdirs = set() # type: Set[str]
for dirname, subdirs, files in os.walk(root):
all_subdirs.update(norm_join(root, dirname, d)
for d in subdirs)
@ -135,6 +149,7 @@ def compress_for_rename(paths):
def compress_for_output_listing(paths):
# type: (Iterable[str]) -> Tuple[Set[str], Set[str]]
"""Returns a tuple of 2 sets of which paths to display to user
The first set contains paths that would be deleted. Files of a package
@ -145,7 +160,7 @@ def compress_for_output_listing(paths):
folders.
"""
will_remove = list(paths)
will_remove = set(paths)
will_skip = set()
# Determine folders and files
@ -158,7 +173,8 @@ def compress_for_output_listing(paths):
folders.add(os.path.dirname(path))
files.add(path)
_normcased_files = set(map(os.path.normcase, files))
# probably this one https://github.com/python/mypy/issues/390
_normcased_files = set(map(os.path.normcase, files)) # type: ignore
folders = compact(folders)
@ -187,21 +203,23 @@ class StashedUninstallPathSet(object):
"""A set of file rename operations to stash files while
tentatively uninstalling them."""
def __init__(self):
# type: () -> None
# Mapping from source file root to [Adjacent]TempDirectory
# for files under that directory.
self._save_dirs = {}
self._save_dirs = {} # type: Dict[str, TempDirectory]
# (old path, new path) tuples for each move that may need
# to be undone.
self._moves = []
self._moves = [] # type: List[Tuple[str, str]]
def _get_directory_stash(self, path):
# type: (str) -> str
"""Stashes a directory.
Directories are stashed adjacent to their original location if
possible, or else moved/copied into the user's temp dir."""
try:
save_dir = AdjacentTempDirectory(path)
save_dir = AdjacentTempDirectory(path) # type: TempDirectory
save_dir.create()
except OSError:
save_dir = TempDirectory(kind="uninstall")
@ -211,6 +229,7 @@ class StashedUninstallPathSet(object):
return save_dir.path
def _get_file_stash(self, path):
# type: (str) -> str
"""Stashes a file.
If no root has been provided, one will be created for the directory
@ -239,6 +258,7 @@ class StashedUninstallPathSet(object):
return save_dir.path
def stash(self, path):
# type: (str) -> str
"""Stashes the directory or file and returns its new location.
"""
if os.path.isdir(path):
@ -258,6 +278,7 @@ class StashedUninstallPathSet(object):
return new_path
def commit(self):
# type: () -> None
"""Commits the uninstall by removing stashed files."""
for _, save_dir in self._save_dirs.items():
save_dir.cleanup()
@ -265,6 +286,7 @@ class StashedUninstallPathSet(object):
self._save_dirs = {}
def rollback(self):
# type: () -> None
"""Undoes the uninstall by moving stashed files back."""
for p in self._moves:
logging.info("Moving to %s\n from %s", *p)
@ -285,6 +307,7 @@ class StashedUninstallPathSet(object):
@property
def can_rollback(self):
# type: () -> bool
return bool(self._moves)
@ -292,13 +315,15 @@ class UninstallPathSet(object):
"""A set of file paths to be removed in the uninstallation of a
requirement."""
def __init__(self, dist):
self.paths = set()
self._refuse = set()
self.pth = {}
# type: (Distribution) -> None
self.paths = set() # type: Set[str]
self._refuse = set() # type: Set[str]
self.pth = {} # type: Dict[str, UninstallPthEntries]
self.dist = dist
self._moved_paths = StashedUninstallPathSet()
def _permitted(self, path):
# type: (str) -> bool
"""
Return True if the given path is one we are permitted to
remove/modify, False otherwise.
@ -307,6 +332,7 @@ class UninstallPathSet(object):
return is_local(path)
def add(self, path):
# type: (str) -> None
head, tail = os.path.split(path)
# we normalize the head to resolve parent directory symlinks, but not
@ -326,6 +352,7 @@ class UninstallPathSet(object):
self.add(cache_from_source(path))
def add_pth(self, pth_file, entry):
# type: (str, str) -> None
pth_file = normalize_path(pth_file)
if self._permitted(pth_file):
if pth_file not in self.pth:
@ -335,6 +362,7 @@ class UninstallPathSet(object):
self._refuse.add(pth_file)
def remove(self, auto_confirm=False, verbose=False):
# type: (bool, bool) -> None
"""Remove paths in ``self.paths`` with confirmation (unless
``auto_confirm`` is True)."""
@ -366,10 +394,12 @@ class UninstallPathSet(object):
logger.info('Successfully uninstalled %s', dist_name_version)
def _allowed_to_proceed(self, verbose):
# type: (bool) -> bool
"""Display which files would be deleted and prompt for confirmation
"""
def _display(msg, paths):
# type: (str, Iterable[str]) -> None
if not paths:
return
@ -383,7 +413,7 @@ class UninstallPathSet(object):
else:
# In verbose mode, display all the files that are going to be
# deleted.
will_remove = list(self.paths)
will_remove = set(self.paths)
will_skip = set()
_display('Would remove:', will_remove)
@ -395,24 +425,27 @@ class UninstallPathSet(object):
return ask('Proceed (y/n)? ', ('y', 'n')) == 'y'
def rollback(self):
# type: () -> None
"""Rollback the changes previously made by remove()."""
if not self._moved_paths.can_rollback:
logger.error(
"Can't roll back %s; was not uninstalled",
self.dist.project_name,
)
return False
return
logger.info('Rolling back uninstall of %s', self.dist.project_name)
self._moved_paths.rollback()
for pth in self.pth.values():
pth.rollback()
def commit(self):
# type: () -> None
"""Remove temporary save dir: rollback will no longer be possible."""
self._moved_paths.commit()
@classmethod
def from_dist(cls, dist):
# type: (Distribution) -> UninstallPathSet
dist_path = normalize_path(dist.location)
if not dist_is_local(dist):
logger.info(
@ -544,15 +577,17 @@ class UninstallPathSet(object):
class UninstallPthEntries(object):
def __init__(self, pth_file):
# type: (str) -> None
if not os.path.isfile(pth_file):
raise UninstallationError(
"Cannot remove entries from nonexistent file %s" % pth_file
)
self.file = pth_file
self.entries = set()
self._saved_lines = None
self.entries = set() # type: Set[str]
self._saved_lines = None # type: Optional[List[bytes]]
def add(self, entry):
# type: (str) -> None
entry = os.path.normcase(entry)
# On Windows, os.path.normcase converts the entry to use
# backslashes. This is correct for entries that describe absolute
@ -563,6 +598,7 @@ class UninstallPthEntries(object):
self.entries.add(entry)
def remove(self):
# type: () -> None
logger.debug('Removing pth entries from %s:', self.file)
with open(self.file, 'rb') as fh:
# windows uses '\r\n' with py3k, but uses '\n' with py2.x
@ -585,6 +621,7 @@ class UninstallPthEntries(object):
fh.writelines(lines)
def rollback(self):
# type: () -> bool
if self._saved_lines is None:
logger.error(
'Cannot roll back changes to %s, none were made', self.file