From 44b3c90bfd8cd584b5de4acbddd608d6343dd1a4 Mon Sep 17 00:00:00 2001 From: Diego Ramirez Date: Sat, 26 Jun 2021 19:36:35 -0500 Subject: [PATCH] Complete type annotations in pip/_internal/cli Co-authored-by: Tzu-ping Chung --- news/10065.trivial.rst | 1 + src/pip/_internal/cli/progress_bars.py | 35 +++----- src/pip/_internal/cli/req_command.py | 109 ++++++++++++------------- src/pip/_internal/cli/spinners.py | 51 ++++-------- 4 files changed, 81 insertions(+), 115 deletions(-) create mode 100644 news/10065.trivial.rst diff --git a/news/10065.trivial.rst b/news/10065.trivial.rst new file mode 100644 index 000000000..d6bbe71c2 --- /dev/null +++ b/news/10065.trivial.rst @@ -0,0 +1 @@ +Fixed all the annotations from ``pip/_internal/cli``. diff --git a/src/pip/_internal/cli/progress_bars.py b/src/pip/_internal/cli/progress_bars.py index 3064c8569..4d1413af3 100644 --- a/src/pip/_internal/cli/progress_bars.py +++ b/src/pip/_internal/cli/progress_bars.py @@ -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 diff --git a/src/pip/_internal/cli/req_command.py b/src/pip/_internal/cli/req_command.py index 7ed623880..377351e20 100644 --- a/src/pip/_internal/cli/req_command.py +++ b/src/pip/_internal/cli/req_command.py @@ -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. diff --git a/src/pip/_internal/cli/spinners.py b/src/pip/_internal/cli/spinners.py index 08e156617..4d3ae6e29 100644 --- a/src/pip/_internal/cli/spinners.py +++ b/src/pip/_internal/cli/spinners.py @@ -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: