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

Complete type annotations in pip/_internal/cli 

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>
This commit is contained in:
Diego Ramirez 2021-06-26 19:36:35 -05:00 committed by GitHub
parent 9c2c52bdbd
commit 44b3c90bfd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 81 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
news/10065.trivial.rst Normal file
View File

@ -0,0 +1 @@
Fixed all the annotations from ``pip/_internal/cli``.

35
src/pip/_internal/cli/progress_bars.py
View File

@ -1,7 +1,7 @@
import itertools
import sys
from signal import SIGINT, default_int_handler, signal
from typing import Any, Dict, List
from typing import Any
from pip._vendor.progress.bar import Bar, FillingCirclesBar, IncrementalBar
from pip._vendor.progress.spinner import Spinner
@ -18,8 +18,7 @@ except Exception:
colorama = None
def _select_progress_class(preferred, fallback):
# type: (Bar, Bar) -> Bar
def _select_progress_class(preferred: Bar, fallback: Bar) -> Bar:
encoding = getattr(preferred.file, "encoding", None)
# If we don't know what encoding this file is in, then we'll just assume
@ -67,8 +66,7 @@ class InterruptibleMixin:
download has already completed, for example.
"""
def __init__(self, *args, **kwargs):
# type: (List[Any], Dict[Any, Any]) -> None
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""
Save the original SIGINT handler for later.
"""
@ -85,8 +83,7 @@ class InterruptibleMixin:
if self.original_handler is None:
self.original_handler = default_int_handler
def finish(self):
# type: () -> None
def finish(self) -> None:
"""
Restore the original SIGINT handler after finishing.
@ -108,8 +105,7 @@ class InterruptibleMixin:
class SilentBar(Bar):
def update(self):
# type: () -> None
def update(self) -> None:
pass
@ -122,28 +118,24 @@ class BlueEmojiBar(IncrementalBar):
class DownloadProgressMixin:
def __init__(self, *args, **kwargs):
# type: (List[Any], Dict[Any, Any]) -> None
def __init__(self, *args: Any, **kwargs: Any) -> None:
# https://github.com/python/mypy/issues/5887
super().__init__(*args, **kwargs) # type: ignore
self.message = (" " * (get_indentation() + 2)) + self.message # type: str
@property
def downloaded(self):
# type: () -> str
def downloaded(self) -> str:
return format_size(self.index) # type: ignore
@property
def download_speed(self):
# type: () -> str
def download_speed(self) -> str:
# Avoid zero division errors...
if self.avg == 0.0: # type: ignore
return "..."
return format_size(1 / self.avg) + "/s" # type: ignore
@property
def pretty_eta(self):
# type: () -> str
def pretty_eta(self) -> str:
if self.eta: # type: ignore
return f"eta {self.eta_td}" # type: ignore
return ""
@ -158,8 +150,7 @@ class DownloadProgressMixin:
class WindowsMixin:
def __init__(self, *args, **kwargs):
# type: (List[Any], Dict[Any, Any]) -> None
def __init__(self, *args: Any, **kwargs: Any) -> None:
# The Windows terminal does not support the hide/show cursor ANSI codes
# even with colorama. So we'll ensure that hide_cursor is False on
# Windows.
@ -221,14 +212,12 @@ class DownloadProgressSpinner(
file = sys.stdout
suffix = "%(downloaded)s %(download_speed)s"
def next_phase(self):
# type: () -> str
def next_phase(self) -> str:
if not hasattr(self, "_phaser"):
self._phaser = itertools.cycle(self.phases)
return next(self._phaser)
def update(self):
# type: () -> None
def update(self) -> None:
message = self.message % self
phase = self.next_phase()
suffix = self.suffix % self

109
src/pip/_internal/cli/req_command.py
View File

@ -50,14 +50,12 @@ class SessionCommandMixin(CommandContextMixIn):
A class mixin for command classes needing _build_session().
"""
def __init__(self):
# type: () -> None
def __init__(self) -> None:
super().__init__()
self._session = None # Optional[PipSession]
self._session: Optional[PipSession] = None
@classmethod
def _get_index_urls(cls, options):
# type: (Values) -> Optional[List[str]]
def _get_index_urls(cls, options: Values) -> Optional[List[str]]:
"""Return a list of index urls from user-provided options."""
index_urls = []
if not getattr(options, "no_index", False):
@ -70,8 +68,7 @@ class SessionCommandMixin(CommandContextMixIn):
# Return None rather than an empty list
return index_urls or None
def get_default_session(self, options):
# type: (Values) -> PipSession
def get_default_session(self, options: Values) -> PipSession:
"""Get a default-managed session."""
if self._session is None:
self._session = self.enter_context(self._build_session(options))
@ -81,8 +78,12 @@ class SessionCommandMixin(CommandContextMixIn):
assert self._session is not None
return self._session
def _build_session(self, options, retries=None, timeout=None):
# type: (Values, Optional[int], Optional[int]) -> PipSession
def _build_session(
self,
options: Values,
retries: Optional[int] = None,
timeout: Optional[int] = None,
) -> PipSession:
assert not options.cache_dir or os.path.isabs(options.cache_dir)
session = PipSession(
cache=(
@ -126,8 +127,7 @@ class IndexGroupCommand(Command, SessionCommandMixin):
This also corresponds to the commands that permit the pip version check.
"""
def handle_pip_version_check(self, options):
# type: (Values) -> None
def handle_pip_version_check(self, options: Values) -> None:
"""
Do the pip version check if not disabled.
@ -154,8 +154,7 @@ KEEPABLE_TEMPDIR_TYPES = [
]
def warn_if_run_as_root():
# type: () -> None
def warn_if_run_as_root() -> None:
"""Output a warning for sudo users on Unix.
In a virtual environment, sudo pip still writes to virtualenv.
@ -184,19 +183,18 @@ def warn_if_run_as_root():
)
def with_cleanup(func):
# type: (Any) -> Any
def with_cleanup(func: Any) -> Any:
"""Decorator for common logic related to managing temporary
directories.
"""
def configure_tempdir_registry(registry):
# type: (TempDirectoryTypeRegistry) -> None
def configure_tempdir_registry(registry: TempDirectoryTypeRegistry) -> None:
for t in KEEPABLE_TEMPDIR_TYPES:
registry.set_delete(t, False)
def wrapper(self, options, args):
# type: (RequirementCommand, Values, List[Any]) -> Optional[int]
def wrapper(
self: RequirementCommand, options: Values, args: List[Any]
) -> Optional[int]:
assert self.tempdir_registry is not None
if options.no_clean:
configure_tempdir_registry(self.tempdir_registry)
@ -214,15 +212,13 @@ def with_cleanup(func):
class RequirementCommand(IndexGroupCommand):
def __init__(self, *args, **kw):
# type: (Any, Any) -> None
def __init__(self, *args: Any, **kw: Any) -> None:
super().__init__(*args, **kw)
self.cmd_opts.add_option(cmdoptions.no_clean())
@staticmethod
def determine_resolver_variant(options):
# type: (Values) -> str
def determine_resolver_variant(options: Values) -> str:
"""Determines which resolver should be used, based on the given options."""
if "legacy-resolver" in options.deprecated_features_enabled:
return "legacy"
@ -232,15 +228,14 @@ class RequirementCommand(IndexGroupCommand):
@classmethod
def make_requirement_preparer(
cls,
temp_build_dir, # type: TempDirectory
options, # type: Values
req_tracker, # type: RequirementTracker
session, # type: PipSession
finder, # type: PackageFinder
use_user_site, # type: bool
download_dir=None, # type: str
):
# type: (...) -> RequirementPreparer
temp_build_dir: TempDirectory,
options: Values,
req_tracker: RequirementTracker,
session: PipSession,
finder: PackageFinder,
use_user_site: bool,
download_dir: Optional[str] = None,
) -> RequirementPreparer:
"""
Create a RequirementPreparer instance for the given parameters.
"""
@ -283,19 +278,18 @@ class RequirementCommand(IndexGroupCommand):
@classmethod
def make_resolver(
cls,
preparer, # type: RequirementPreparer
finder, # type: PackageFinder
options, # type: Values
wheel_cache=None, # type: Optional[WheelCache]
use_user_site=False, # type: bool
ignore_installed=True, # type: bool
ignore_requires_python=False, # type: bool
force_reinstall=False, # type: bool
upgrade_strategy="to-satisfy-only", # type: str
use_pep517=None, # type: Optional[bool]
py_version_info=None, # type: Optional[Tuple[int, ...]]
):
# type: (...) -> BaseResolver
preparer: RequirementPreparer,
finder: PackageFinder,
options: Values,
wheel_cache: Optional[WheelCache] = None,
use_user_site: bool = False,
ignore_installed: bool = True,
ignore_requires_python: bool = False,
force_reinstall: bool = False,
upgrade_strategy: str = "to-satisfy-only",
use_pep517: Optional[bool] = None,
py_version_info: Optional[Tuple[int, ...]] = None,
) -> BaseResolver:
"""
Create a Resolver instance for the given parameters.
"""
@ -342,12 +336,11 @@ class RequirementCommand(IndexGroupCommand):
def get_requirements(
self,
args, # type: List[str]
options, # type: Values
finder, # type: PackageFinder
session, # type: PipSession
):
# type: (...) -> List[InstallRequirement]
args: List[str],
options: Values,
finder: PackageFinder,
session: PipSession,
) -> List[InstallRequirement]:
"""
Parse command-line arguments into the corresponding requirements.
"""
@ -421,8 +414,7 @@ class RequirementCommand(IndexGroupCommand):
return requirements
@staticmethod
def trace_basic_info(finder):
# type: (PackageFinder) -> None
def trace_basic_info(finder: PackageFinder) -> None:
"""
Trace basic information about the provided objects.
"""
@ -434,12 +426,11 @@ class RequirementCommand(IndexGroupCommand):
def _build_package_finder(
self,
options, # type: Values
session, # type: PipSession
target_python=None, # type: Optional[TargetPython]
ignore_requires_python=None, # type: Optional[bool]
):
# type: (...) -> PackageFinder
options: Values,
session: PipSession,
target_python: Optional[TargetPython] = None,
ignore_requires_python: Optional[bool] = None,
) -> PackageFinder:
"""
Create a package finder appropriate to this requirement command.

51
src/pip/_internal/cli/spinners.py
View File

@ -14,25 +14,22 @@ logger = logging.getLogger(__name__)
class SpinnerInterface:
def spin(self):
# type: () -> None
def spin(self) -> None:
raise NotImplementedError()
def finish(self, final_status):
# type: (str) -> None
def finish(self, final_status: str) -> None:
raise NotImplementedError()
class InteractiveSpinner(SpinnerInterface):
def __init__(
self,
message,
file=None,
spin_chars="-\\|/",
message: str,
file: IO[str] = None,
spin_chars: str = "-\\|/",
# Empirically, 8 updates/second looks nice
min_update_interval_seconds=0.125,
min_update_interval_seconds: float = 0.125,
):
# type: (str, IO[str], str, float) -> None
self._message = message
if file is None:
file = sys.stdout
@ -45,8 +42,7 @@ class InteractiveSpinner(SpinnerInterface):
self._file.write(" " * get_indentation() + self._message + " ... ")
self._width = 0
def _write(self, status):
# type: (str) -> None
def _write(self, status: str) -> None:
assert not self._finished
# Erase what we wrote before by backspacing to the beginning, writing
# spaces to overwrite the old text, and then backspacing again
@ -58,16 +54,14 @@ class InteractiveSpinner(SpinnerInterface):
self._file.flush()
self._rate_limiter.reset()
def spin(self):
# type: () -> None
def spin(self) -> None:
if self._finished:
return
if not self._rate_limiter.ready():
return
self._write(next(self._spin_cycle))
def finish(self, final_status):
# type: (str) -> None
def finish(self, final_status: str) -> None:
if self._finished:
return
self._write(final_status)
@ -81,29 +75,25 @@ class InteractiveSpinner(SpinnerInterface):
# act as a keep-alive for systems like Travis-CI that take lack-of-output as
# an indication that a task has frozen.
class NonInteractiveSpinner(SpinnerInterface):
def __init__(self, message, min_update_interval_seconds=60):
# type: (str, float) -> None
def __init__(self, message: str, min_update_interval_seconds: float = 60.0) -> None:
self._message = message
self._finished = False
self._rate_limiter = RateLimiter(min_update_interval_seconds)
self._update("started")
def _update(self, status):
# type: (str) -> None
def _update(self, status: str) -> None:
assert not self._finished
self._rate_limiter.reset()
logger.info("%s: %s", self._message, status)
def spin(self):
# type: () -> None
def spin(self) -> None:
if self._finished:
return
if not self._rate_limiter.ready():
return
self._update("still running...")
def finish(self, final_status):
# type: (str) -> None
def finish(self, final_status: str) -> None:
if self._finished:
return
self._update(f"finished with status '{final_status}'")
@ -111,25 +101,21 @@ class NonInteractiveSpinner(SpinnerInterface):
class RateLimiter:
def __init__(self, min_update_interval_seconds):
# type: (float) -> None
def __init__(self, min_update_interval_seconds: float) -> None:
self._min_update_interval_seconds = min_update_interval_seconds
self._last_update = 0 # type: float
def ready(self):
# type: () -> bool
def ready(self) -> bool:
now = time.time()
delta = now - self._last_update
return delta >= self._min_update_interval_seconds
def reset(self):
# type: () -> None
def reset(self) -> None:
self._last_update = time.time()
@contextlib.contextmanager
def open_spinner(message):
# type: (str) -> Iterator[SpinnerInterface]
def open_spinner(message: str) -> Iterator[SpinnerInterface]:
# Interactive spinner goes directly to sys.stdout rather than being routed
# through the logging system, but it acts like it has level INFO,
# i.e. it's only displayed if we're at level INFO or better.
@ -153,8 +139,7 @@ def open_spinner(message):
@contextlib.contextmanager
def hidden_cursor(file):
# type: (IO[str]) -> Iterator[None]
def hidden_cursor(file: IO[str]) -> Iterator[None]:
# The Windows terminal does not support the hide/show cursor ANSI codes,
# even via colorama. So don't even try.
if WINDOWS: