Skip to content

Commit

Permalink
add mypy annotations for progress_bars and spinners
Browse files Browse the repository at this point in the history
  • Loading branch information
sinscary committed Feb 20, 2020
1 parent b7f3c40 commit fe1d58b
Show file tree
Hide file tree
Showing 2 changed files with 35 additions and 20 deletions.
48 changes: 30 additions & 18 deletions src/pip/_internal/cli/progress_bars.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import itertools
import sys
from signal import SIGINT, default_int_handler, signal
from signal import SIGINT, default_int_handler, signal, Signals

from pip._vendor import six
from pip._vendor.progress.bar import Bar, FillingCirclesBar, IncrementalBar
Expand All @@ -14,7 +14,8 @@
from pip._internal.utils.typing import MYPY_CHECK_RUNNING

if MYPY_CHECK_RUNNING:
from typing import Any
from typing import Any, Dict, List, Iterator
from types import FrameType

try:
from pip._vendor import colorama
Expand All @@ -25,6 +26,7 @@


def _select_progress_class(preferred, fallback):
# type: (Bar, Bar) -> Bar
encoding = getattr(preferred.file, "encoding", None)

# If we don't know what encoding this file is in, then we'll just assume
Expand Down Expand Up @@ -73,10 +75,11 @@ class InterruptibleMixin(object):
"""

def __init__(self, *args, **kwargs):
# type: (List[Any], Dict[Any, Any]) -> None
"""
Save the original SIGINT handler for later.
"""
super(InterruptibleMixin, self).__init__(*args, **kwargs)
super(InterruptibleMixin, self).__init__(*args, **kwargs) # type: ignore

self.original_handler = signal(SIGINT, self.handle_sigint)

Expand All @@ -89,29 +92,32 @@ def __init__(self, *args, **kwargs):
self.original_handler = default_int_handler

def finish(self):
# type: () -> None
"""
Restore the original SIGINT handler after finishing.
This should happen regardless of whether the progress display finishes
normally, or gets interrupted.
"""
super(InterruptibleMixin, self).finish()
super(InterruptibleMixin, self).finish() # type: ignore
signal(SIGINT, self.original_handler)

def handle_sigint(self, signum, frame):
# type: (Signals, FrameType) -> None
"""
Call self.finish() before delegating to the original SIGINT handler.
This handler should only be in place while the progress display is
active.
"""
self.finish()
self.original_handler(signum, frame)
self.original_handler(signum, frame) # type: ignore


class SilentBar(Bar):

def update(self):
# type: () -> None
pass


Expand All @@ -126,27 +132,31 @@ class BlueEmojiBar(IncrementalBar):
class DownloadProgressMixin(object):

def __init__(self, *args, **kwargs):
super(DownloadProgressMixin, self).__init__(*args, **kwargs)
self.message = (" " * (get_indentation() + 2)) + self.message
# type: (List[Any], Dict[Any, Any]) -> None
super(DownloadProgressMixin, self).__init__(*args, **kwargs) # type: ignore
self.message = (" " * (get_indentation() + 2)) + self.message # type: str

@property
def downloaded(self):
return format_size(self.index)
# type: () -> str
return format_size(self.index) # type: ignore

@property
def download_speed(self):
# type: () -> str
# Avoid zero division errors...
if self.avg == 0.0:
if self.avg == 0.0: # type: ignore
return "..."
return format_size(1 / self.avg) + "/s"
return format_size(1 / self.avg) + "/s" # type: ignore

@property
def pretty_eta(self):
if self.eta:
return "eta {}".format(self.eta_td)
# type: () -> str
if self.eta: # type: ignore
return "eta {}".format(self.eta_td) # type: ignore
return ""

def iter(self, it):
def iter(self, it): # type: ignore
for x in it:
yield x
self.next(len(x))
Expand All @@ -156,22 +166,23 @@ def iter(self, it):
class WindowsMixin(object):

def __init__(self, *args, **kwargs):
# type: (List[Any], Dict[Any, 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.
# This call needs to go before the super() call, so that hide_cursor
# is set in time. The base progress bar class writes the "hide cursor"
# code to the terminal in its init, so if we don't set this soon
# enough, we get a "hide" with no corresponding "show"...
if WINDOWS and self.hide_cursor:
if WINDOWS and self.hide_cursor: # type: ignore
self.hide_cursor = False

super(WindowsMixin, self).__init__(*args, **kwargs)
super(WindowsMixin, self).__init__(*args, **kwargs) # type: ignore

# Check if we are running on Windows and we have the colorama module,
# if we do then wrap our file with it.
if WINDOWS and colorama:
self.file = colorama.AnsiToWin32(self.file)
self.file = colorama.AnsiToWin32(self.file) # type: ignore
# The progress code expects to be able to call self.file.isatty()
# but the colorama.AnsiToWin32() object doesn't have that, so we'll
# add it.
Expand Down Expand Up @@ -223,12 +234,13 @@ class DownloadProgressSpinner(WindowsMixin, InterruptibleMixin,
file = sys.stdout
suffix = "%(downloaded)s %(download_speed)s"

def next_phase(self):
def next_phase(self): # type: ignore
if not hasattr(self, "_phaser"):
self._phaser = itertools.cycle(self.phases)
return next(self._phaser)

def update(self):
# type: () -> None
message = self.message % self
phase = self.next_phase()
suffix = self.suffix % self
Expand All @@ -252,7 +264,7 @@ def update(self):
}


def DownloadProgressProvider(progress_bar, max=None):
def DownloadProgressProvider(progress_bar, max=None): # type: ignore
if max is None or max == 0:
return BAR_TYPES[progress_bar][1]().iter
else:
Expand Down
7 changes: 5 additions & 2 deletions src/pip/_internal/cli/spinners.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from pip._internal.utils.typing import MYPY_CHECK_RUNNING

if MYPY_CHECK_RUNNING:
from typing import Iterator, IO
from typing import Iterator, IO, Any

logger = logging.getLogger(__name__)

Expand All @@ -32,6 +32,7 @@ class InteractiveSpinner(SpinnerInterface):
def __init__(self, message, file=None, spin_chars="-\\|/",
# Empirically, 8 updates/second looks nice
min_update_interval_seconds=0.125):
# type: (str, IO[str], str, float) -> None
self._message = message
if file is None:
file = sys.stdout
Expand All @@ -45,6 +46,7 @@ def __init__(self, message, file=None, spin_chars="-\\|/",
self._width = 0

def _write(self, status):
# type: (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
Expand Down Expand Up @@ -87,6 +89,7 @@ def __init__(self, message, min_update_interval_seconds=60):
self._update("started")

def _update(self, status):
# type: (str) -> None
assert not self._finished
self._rate_limiter.reset()
logger.info("%s: %s", self._message, status)
Expand Down Expand Up @@ -151,7 +154,7 @@ def open_spinner(message):

@contextlib.contextmanager
def hidden_cursor(file):
# type: (IO) -> Iterator[None]
# type: (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:
Expand Down

0 comments on commit fe1d58b

Please sign in to comment.