diff --git a/progressbar/__init__.py b/progressbar/__init__.py index a223c0d..3c998a9 100644 --- a/progressbar/__init__.py +++ b/progressbar/__init__.py @@ -141,6 +141,10 @@ def __dir__() -> list[str]: __date__ = str(date.today()) +#: Canonical export list, kept equal to ``sorted(_NAME_TO_MODULE)`` (the single +#: source of truth for lazily re-exported names) plus the eagerly imported +#: dunders. Held as a static literal so type checkers can see the re-exports; +#: ``tests/test_init_exports.py`` asserts it stays in sync with the mapping. __all__ = [ 'ETA', 'AbsoluteETA', diff --git a/progressbar/bar.py b/progressbar/bar.py index ab996f7..ed35a72 100644 --- a/progressbar/bar.py +++ b/progressbar/bar.py @@ -1,6 +1,7 @@ from __future__ import annotations import abc +import collections.abc import contextlib import functools import importlib @@ -18,7 +19,7 @@ from datetime import datetime from types import FrameType -from python_utils import converters, types +from python_utils import converters import progressbar.env import progressbar.terminal @@ -63,15 +64,15 @@ def _load_widgets() -> typing.Any: # float also accepts integers and longs but we don't want an explicit union # due to type checking complexity NumberT = float -ValueT = typing.Union[NumberT, type[base.UnknownLength], None] +ValueT = NumberT | type[base.UnknownLength] | None -T = types.TypeVar('T') +T = typing.TypeVar('T') class ProgressBarMixinBase(abc.ABC): _started = False _finished = False - _last_update_time: types.Optional[float] = None + _last_update_time: float | None = None #: The terminal width. This should be automatically detected but will #: fall back to 80 if auto detection is not possible. @@ -79,28 +80,28 @@ class ProgressBarMixinBase(abc.ABC): #: The widgets to render, defaults to the result of `default_widget()` #: (typed loosely as Any to avoid a static bar->widgets import cycle; the #: public ``progressbar()`` shortcut keeps the precise WidgetBase typing). - widgets: types.MutableSequence[typing.Any] + widgets: collections.abc.MutableSequence[typing.Any] #: When going beyond the max_value, raise an error if True or silently #: ignore otherwise max_error: bool #: Prefix the progressbar with the given string - prefix: types.Optional[str] + prefix: str | None #: Suffix the progressbar with the given string - suffix: types.Optional[str] + suffix: str | None #: Justify to the left if `True` or the right if `False` left_justify: bool #: The default keyword arguments for the `default_widgets` if no widgets #: are configured - widget_kwargs: types.Dict[str, types.Any] + widget_kwargs: dict[str, typing.Any] #: Custom length function for multibyte characters such as CJK # mypy and pyright can't agree on what the correct one is... so we'll # need to use a helper function :( # custom_len: types.Callable[['ProgressBarMixinBase', str], int] - custom_len: types.Callable[[str], int] + custom_len: collections.abc.Callable[[str], int] #: The time the progress bar was started - initial_start_time: types.Optional[datetime] + initial_start_time: datetime | None #: The interval to poll for updates in seconds if there are updates - poll_interval: types.Optional[float] + poll_interval: float | None #: The minimum interval to poll for updates in seconds even if there are #: no updates min_poll_interval: float @@ -115,10 +116,10 @@ class ProgressBarMixinBase(abc.ABC): #: Current progress (min_value <= value <= max_value) value: NumberT #: Previous progress value - previous_value: types.Optional[NumberT] + previous_value: NumberT | None #: Value at the last actual redraw (internal; used by the update gate's #: pixel check, kept separate from the public `previous_value`) - _last_drawn_value: types.Optional[NumberT] + _last_drawn_value: NumberT | None #: The minimum/start value for the progress bar min_value: NumberT #: Maximum (and final) value. Beyond this value an error will be raised @@ -126,24 +127,24 @@ class ProgressBarMixinBase(abc.ABC): max_value: ValueT #: The time the progressbar reached `max_value` or when `finish()` was #: called. - end_time: types.Optional[datetime] + end_time: datetime | None #: The time `start()` was called or iteration started. - start_time: types.Optional[datetime] + start_time: datetime | None #: Seconds between `start_time` and last call to `update()` seconds_elapsed: float #: Extra data for widgets with persistent state. This is used by #: sampling widgets for example. Since widgets can be shared between #: multiple progressbars we need to store the state with the progressbar. - extra: types.Dict[str, types.Any] + extra: dict[str, typing.Any] - def get_last_update_time(self) -> types.Optional[datetime]: + def get_last_update_time(self) -> datetime | None: if self._last_update_time: return datetime.fromtimestamp(self._last_update_time) else: return None - def set_last_update_time(self, value: types.Optional[datetime]): + def set_last_update_time(self, value: datetime | None): if value: self._last_update_time = time.mktime(value.timetuple()) else: @@ -177,7 +178,7 @@ def __del__(self): def __getstate__(self): return self.__dict__ - def data(self) -> types.Dict[str, types.Any]: # pragma: no cover + def data(self) -> dict[str, typing.Any]: # pragma: no cover raise NotImplementedError() def started(self) -> bool: @@ -187,7 +188,7 @@ def finished(self) -> bool: return self._finished -class ProgressBarBase(types.Iterable[NumberT], ProgressBarMixinBase): +class ProgressBarBase(collections.abc.Iterable[NumberT], ProgressBarMixinBase): _index_counter = itertools.count() index: int = -1 label: str = '' @@ -206,7 +207,6 @@ def __repr__(self): label = f': {self.label}' if self.label else '' return f'<{self.__class__.__name__}#{self.index}{label}>' - class DefaultFdMixin(ProgressBarMixinBase): # The file descriptor to write to. Defaults to `sys.stderr` fd: base.TextIO = sys.stderr @@ -337,14 +337,14 @@ def _determine_enable_colors( return color_support - def print(self, *args: types.Any, **kwargs: types.Any) -> None: + def print(self, *args: typing.Any, **kwargs: typing.Any) -> None: print(*args, file=self.fd, **kwargs) def start(self, **kwargs: typing.Any): os_specific.set_console_mode() super().start(**kwargs) - def update(self, *args: types.Any, **kwargs: types.Any) -> None: + def update(self, *args: typing.Any, **kwargs: typing.Any) -> None: super().update(*args, **kwargs) line: str = converters.to_unicode(self._format_line()) @@ -356,12 +356,15 @@ def update(self, *args: types.Any, **kwargs: types.Any) -> None: try: # pragma: no cover self.fd.write(line) except UnicodeEncodeError: # pragma: no cover - self.fd.write(types.cast(str, line.encode('ascii', 'replace'))) + # ``fd`` is a text stream, so write an ASCII-safe *str*: encode + # with 'replace' to drop un-encodable characters, then decode + # back. Writing the raw bytes here would raise ``TypeError``. + self.fd.write(line.encode('ascii', 'replace').decode('ascii')) def finish( self, - *args: types.Any, - **kwargs: types.Any, + *args: typing.Any, + **kwargs: typing.Any, ) -> None: # pragma: no cover os_specific.reset_console_mode() @@ -564,7 +567,7 @@ def start(self, *args: typing.Any, **kwargs: typing.Any): utils.streams.start_capturing(self) super().start(*args, **kwargs) - def update(self, value: types.Optional[NumberT] = None): + def update(self, value: NumberT | None = None): cleared = not self.line_breaks and utils.streams.needs_clear() if cleared: self.fd.write('\r' + ' ' * self.term_width + '\r') @@ -661,24 +664,24 @@ class ProgressBar( you from changing the ProgressBar you should treat it as read only. """ - _iterable: types.Optional[types.Iterator] + _iterable: collections.abc.Iterator | None _DEFAULT_MAXVAL: type[base.UnknownLength] = base.UnknownLength # update every 50 milliseconds (up to a 20 times per second) _MINIMUM_UPDATE_INTERVAL: float = 0.050 - _last_update_time: types.Optional[float] = None + _last_update_time: float | None = None paused: bool = False def __init__( self, min_value: NumberT = 0, max_value: ValueT = None, - widgets: types.Optional[types.Sequence[typing.Any]] = None, + widgets: collections.abc.Sequence[typing.Any] | None = None, left_justify: bool = True, initial_value: NumberT = 0, - poll_interval: types.Optional[float] = None, - widget_kwargs: types.Optional[types.Dict[str, types.Any]] = None, - custom_len: types.Callable[[str], int] = utils.len_color, + poll_interval: float | None = None, + widget_kwargs: dict[str, typing.Any] | None = None, + custom_len: collections.abc.Callable[[str], int] = utils.len_color, max_error=True, prefix=None, suffix=None, @@ -693,7 +696,7 @@ def __init__( max_value, poll_interval, kwargs ) - if max_value and min_value > types.cast(NumberT, max_value): + if max_value and min_value > typing.cast(NumberT, max_value): raise ValueError( 'Max value needs to be bigger than the min value', ) @@ -721,9 +724,9 @@ def __init__( def _apply_deprecated_aliases( self, max_value: ValueT, - poll_interval: types.Optional[float], - kwargs: types.Dict[str, typing.Any], - ) -> tuple[ValueT, types.Optional[float]]: + poll_interval: float | None, + kwargs: dict[str, typing.Any], + ) -> tuple[ValueT, float | None]: """Resolve the deprecated ``maxval``/``poll`` keyword aliases. Emits a :py:class:`DeprecationWarning` for each legacy name that is @@ -751,7 +754,7 @@ def _apply_deprecated_aliases( return max_value, poll_interval def _copy_widgets( - self, widgets: types.Optional[types.Sequence[typing.Any]] + self, widgets: collections.abc.Sequence[typing.Any] | None ) -> list[typing.Any]: """Return a fresh widget list, deep-copying the copy-safe widgets. @@ -767,8 +770,8 @@ def _copy_widgets( def _setup_poll_intervals( self, - poll_interval: types.Optional[float], - min_poll_interval: types.Optional[float], + poll_interval: float | None, + min_poll_interval: float | None, ) -> None: """Convert the poll intervals to seconds and clamp the minimum. @@ -799,9 +802,7 @@ def _setup_poll_intervals( float(os.environ.get('PROGRESSBAR_MINIMUM_UPDATE_INTERVAL', 0)), ) # type: ignore - def _seed_variables( - self, variables: types.Optional[types.Dict[str, typing.Any]] - ) -> None: + def _seed_variables(self, variables: dict[str, typing.Any] | None) -> None: """Seed the user-defined variables dict and scan widgets for names. Builds the ``variables`` mapping used by ``Variable``/``FormatWidget`` @@ -900,7 +901,7 @@ def percentage(self) -> float | None: return percentage - def data(self) -> types.Dict[str, types.Any]: + def data(self) -> dict[str, typing.Any]: """ Returns: diff --git a/progressbar/fast.py b/progressbar/fast.py index 46e42aa..56bd4a0 100644 --- a/progressbar/fast.py +++ b/progressbar/fast.py @@ -1,6 +1,7 @@ from __future__ import annotations import typing +from collections.abc import Callable from datetime import datetime, timedelta from . import ( @@ -14,7 +15,7 @@ #: ``speedups`` package — or any caller — swap in a faster/custom formatter. #: This is a supported extension point, exercised by #: ``test_fast_format_line_uses_native_hook``. -_format_fast_line: typing.Callable[[FastProgressBar], str] | None = None +_format_fast_line: Callable[[FastProgressBar], str] | None = None #: Spinner frames cycled for unknown-length bars: bar, forward slash, dash, #: back slash. A plain (non-raw) literal so the escape is a single ``\`` and diff --git a/progressbar/multi.py b/progressbar/multi.py index 254c132..8f1a0c2 100644 --- a/progressbar/multi.py +++ b/progressbar/multi.py @@ -1,5 +1,6 @@ from __future__ import annotations +import collections.abc import enum import importlib import io @@ -26,7 +27,7 @@ # thread and race MultiBar._label_bar's ``assert bar.widgets``. importlib.import_module('progressbar.widgets') -SortKeyFunc = typing.Callable[[bar.ProgressBar], typing.Any] +SortKeyFunc = collections.abc.Callable[[bar.ProgressBar], typing.Any] class _Update(typing.Protocol): @@ -105,7 +106,8 @@ class MultiBar(dict[str, bar.ProgressBar]): def __init__( self, - bars: typing.Iterable[tuple[str, bar.ProgressBar]] | None = None, + bars: collections.abc.Iterable[tuple[str, bar.ProgressBar]] + | None = None, fd: typing.TextIO = sys.stderr, prepend_label: bool = True, append_label: bool = False, @@ -161,7 +163,7 @@ def __init__( super().__init__(bars or {}) - def __setitem__(self, key: str, bar: bar.ProgressBar): + def __setitem__(self, key: str, bar: bar.ProgressBar) -> None: """Add a progressbar to the multibar.""" if bar.label != key or not key: # pragma: no branch bar.label = key @@ -186,7 +188,7 @@ def __delitem__(self, key: str) -> None: self._finished_at.pop(bar_, None) self._labeled.discard(bar_) - def __getitem__(self, key: str): + def __getitem__(self, key: str) -> bar.ProgressBar: """Get (and create if needed) a progressbar from the multibar.""" try: return super().__getitem__(key) @@ -265,7 +267,7 @@ def _render_bar( bar_: bar.ProgressBar, now: float, expired: float | None, - ) -> typing.Iterable[str]: + ) -> collections.abc.Iterable[str]: def update( force: bool = True, write: bool = True ) -> str: # pragma: no cover @@ -294,7 +296,7 @@ def _render_finished_bar( now: float, expired: float | None, update: _Update, - ) -> typing.Iterable[str]: + ) -> collections.abc.Iterable[str]: if bar_ not in self._finished_at: self._finished_at[bar_] = now # Force update to get the finished format @@ -411,17 +413,17 @@ def join(self, timeout: float | None = None) -> None: if not self._thread.is_alive(): self._thread = None - def stop(self, timeout: float | None = None): + def stop(self, timeout: float | None = None) -> None: self._thread_finished.set() self.join(timeout=timeout) - def get_sorted_bars(self): + def get_sorted_bars(self) -> list[bar.ProgressBar]: # Materialize the values into a list first so other threads can # add or remove bars while we are sorting and rendering bars = list(self.values()) return sorted(bars, key=self.sort_keyfunc, reverse=self.sort_reverse) - def __enter__(self): + def __enter__(self) -> MultiBar: self.start() return self diff --git a/progressbar/shortcuts.py b/progressbar/shortcuts.py index c4d2c5d..df72eba 100644 --- a/progressbar/shortcuts.py +++ b/progressbar/shortcuts.py @@ -1,5 +1,6 @@ from __future__ import annotations +import collections.abc import os import typing @@ -15,15 +16,16 @@ def progressbar( - iterator: typing.Iterator[T], + iterator: collections.abc.Iterable[T], min_value: bar.NumberT = 0, max_value: bar.ValueT = None, - widgets: typing.Sequence[widgets_module.WidgetBase | str] | None = None, + widgets: collections.abc.Sequence[widgets_module.WidgetBase | str] + | None = None, prefix: str | None = None, suffix: str | None = None, fast: bool | None = None, **kwargs: typing.Any, -) -> typing.Iterator[T]: +) -> collections.abc.Iterator[T]: # Auto-dispatch to the lean FastProgressBar for the simple, common case; # anything that needs the full widget machinery uses ProgressBar. use_fast = ( diff --git a/progressbar/terminal/base.py b/progressbar/terminal/base.py index 8a79fbc..88eb51d 100644 --- a/progressbar/terminal/base.py +++ b/progressbar/terminal/base.py @@ -1,7 +1,7 @@ from __future__ import annotations import abc -import collections +import collections.abc import colorsys import enum import threading @@ -12,7 +12,7 @@ # `types` module from typing import ClassVar -from python_utils import converters, types +from python_utils import converters from .. import ( base as pbase, @@ -169,9 +169,9 @@ def __call__(self, stream: typing.IO[str]) -> tuple[int, int]: ) if len(res_list) == 1: - return types.cast(types.Tuple[int, int], res_list[0]) + return typing.cast(tuple[int, int], res_list[0]) - return types.cast(types.Tuple[int, int], tuple(res_list)) + return typing.cast(tuple[int, int], tuple(res_list)) def row(self, stream: typing.IO[str]) -> int: row, _ = self(stream) @@ -201,7 +201,7 @@ class WindowsColors(enum.Enum): INTENSE_WHITE = 255, 255, 255 @staticmethod - def from_rgb(rgb: types.Tuple[int, int, int]) -> WindowsColors: + def from_rgb(rgb: tuple[int, int, int]) -> WindowsColors: """ Find the closest WindowsColors to the given RGB color. @@ -225,7 +225,9 @@ def color_distance( rgb1: tuple[int, int, int], rgb2: tuple[int, int, int], ): - return sum((c1 - c2) ** 2 for c1, c2 in zip(rgb1, rgb2)) + return sum( + (c1 - c2) ** 2 for c1, c2 in zip(rgb1, rgb2, strict=False) + ) return min( WindowsColors, @@ -401,7 +403,7 @@ def underline(self) -> DummyColor | SGRColor: return SGRColor(self, 58, 59) @property - def ansi(self) -> types.Optional[str]: + def ansi(self) -> str | None: if ( env.COLOR_SUPPORT is env.ColorSupport.XTERM_TRUECOLOR ): # pragma: no branch @@ -448,30 +450,20 @@ def __hash__(self) -> int: class Colors: - by_name: ClassVar[defaultdict[str, types.List[Color]]] = ( - collections.defaultdict(list) - ) - by_lowername: ClassVar[defaultdict[str, types.List[Color]]] = ( - collections.defaultdict(list) - ) - by_hex: ClassVar[defaultdict[str, types.List[Color]]] = ( - collections.defaultdict(list) - ) - by_rgb: ClassVar[defaultdict[RGB, types.List[Color]]] = ( - collections.defaultdict(list) - ) - by_hls: ClassVar[defaultdict[HSL, types.List[Color]]] = ( - collections.defaultdict(list) - ) + by_name: ClassVar[defaultdict[str, list[Color]]] = defaultdict(list) + by_lowername: ClassVar[defaultdict[str, list[Color]]] = defaultdict(list) + by_hex: ClassVar[defaultdict[str, list[Color]]] = defaultdict(list) + by_rgb: ClassVar[defaultdict[RGB, list[Color]]] = defaultdict(list) + by_hls: ClassVar[defaultdict[HSL, list[Color]]] = defaultdict(list) by_xterm: ClassVar[dict[int, Color]] = dict() @classmethod def register( cls, rgb: RGB, - hls: types.Optional[HSL] = None, - name: types.Optional[str] = None, - xterm: types.Optional[int] = None, + hls: HSL | None = None, + name: str | None = None, + xterm: int | None = None, ) -> Color: if hls is None: hls = HSL.from_rgb(rgb) @@ -497,14 +489,14 @@ def interpolate(cls, color_a: Color, color_b: Color, step: float) -> Color: class ColorGradient: - interpolate: typing.Callable[[Color, Color, float], Color] | None + interpolate: collections.abc.Callable[[Color, Color, float], Color] | None colors: tuple[Color, ...] def __init__( self, *colors: Color, interpolate: ( - typing.Callable[[Color, Color, float], Color] | None + collections.abc.Callable[[Color, Color, float], Color] | None ) = Colors.interpolate, ) -> None: assert colors @@ -555,7 +547,7 @@ def get_color(self, value: float) -> Color: return color -OptionalColor = types.Union[Color, ColorGradient, None] +OptionalColor = Color | ColorGradient | None def get_color(value: float, color: OptionalColor) -> Color | None: @@ -572,7 +564,7 @@ def apply_colors( bg: OptionalColor = None, fg_none: Color | None = None, bg_none: Color | None = None, - **kwargs: types.Any, + **kwargs: typing.Any, ) -> str: """Apply colors/gradients to a string depending on the given percentage. diff --git a/progressbar/terminal/stream.py b/progressbar/terminal/stream.py index 04429e4..707f048 100644 --- a/progressbar/terminal/stream.py +++ b/progressbar/terminal/stream.py @@ -2,7 +2,7 @@ import sys import typing -from collections.abc import Iterable, Iterator +from collections.abc import Generator, Iterable, Iterator from types import TracebackType from progressbar import base @@ -133,7 +133,7 @@ def truncate(self, __size: int | None = None) -> int: return len(self.line) - def __iter__(self) -> typing.Generator[str, typing.Any, typing.Any]: + def __iter__(self) -> Generator[str, typing.Any, typing.Any]: yield self.line def writelines(self, __lines: Iterable[str]) -> None: diff --git a/progressbar/utils.py b/progressbar/utils.py index 3cfc14f..ae7e612 100644 --- a/progressbar/utils.py +++ b/progressbar/utils.py @@ -8,7 +8,8 @@ import os import re import sys -from collections.abc import Iterable, Iterator +import typing +from collections.abc import Callable, Iterable, Iterator from types import TracebackType from python_utils import types @@ -18,9 +19,6 @@ from progressbar import base, env, terminal -if types.TYPE_CHECKING: - from .bar import ProgressBar, ProgressBarMixinBase - # Make sure these are available for import assert timedelta_to_seconds is not None assert get_terminal_size is not None @@ -28,7 +26,22 @@ assert scale_1024 is not None assert epoch is not None -StringT = types.TypeVar('StringT', bound=types.StringTypes) +StringT = typing.TypeVar('StringT', bound=types.StringTypes) +T = typing.TypeVar('T') + + +class _ProgressListener(typing.Protocol): + """Structural type for the bars the stream wrapper notifies. + + Defined locally instead of importing ``ProgressBarMixinBase`` from + ``bar`` so ``utils`` has no dependency on ``bar`` — not even a + type-checking-only one, which CodeQL still reports as a ``bar`` <-> + ``utils`` module-level import cycle. + """ + + def update(self) -> None: + """Redraw in response to redirected output being written.""" + # Precompiled ANSI CSI escape-sequence patterns (str and bytes). Compiled once # at import instead of per no_color() call, which runs for every widget on @@ -39,10 +52,26 @@ ) +@typing.overload +def deltas_to_seconds( + *deltas: None | datetime.timedelta | float | int, + default: type[ValueError] = ..., +) -> float: + """Coalesce to seconds; raise ``ValueError`` if no delta is valid.""" + + +@typing.overload +def deltas_to_seconds( + *deltas: None | datetime.timedelta | float | int, + default: T, +) -> float | T: + """Coalesce to seconds; return ``default`` if no delta is valid.""" + + def deltas_to_seconds( - *deltas: None | datetime.timedelta | float, - default: types.Optional[types.Type[ValueError]] = ValueError, -) -> int | float | None: + *deltas: None | datetime.timedelta | float | int, + default: typing.Any = ValueError, +) -> typing.Any: """ Convert timedeltas and seconds as int to seconds as float while coalescing. @@ -80,8 +109,7 @@ def deltas_to_seconds( if default is ValueError: raise ValueError('No valid deltas passed to `deltas_to_seconds`') else: - # mypy doesn't understand the `default is ValueError` check - return default # type: ignore + return default def no_color(value: StringT) -> StringT: @@ -133,14 +161,14 @@ class WrappingIO: buffer: io.StringIO target: base.IO capturing: bool - listeners: set + listeners: set[_ProgressListener] needs_clear: bool = False def __init__( self, target: base.IO, capturing: bool = False, - listeners: types.Optional[types.Set[ProgressBar]] = None, + listeners: set[_ProgressListener] | None = None, ) -> None: self.buffer = io.StringIO() self.target = target @@ -214,7 +242,7 @@ def seekable(self) -> bool: def tell(self) -> int: return self.target.tell() - def truncate(self, size: types.Optional[int] = None) -> int: + def truncate(self, size: int | None = None) -> int: return self.target.truncate(size) def writable(self) -> bool: @@ -247,9 +275,9 @@ class StreamWrapper: stdout: base.TextIO | WrappingIO stderr: base.TextIO | WrappingIO - original_excepthook: types.Callable[ + original_excepthook: Callable[ [ - types.Type[BaseException], + type[BaseException], BaseException, TracebackType | None, ], @@ -259,7 +287,7 @@ class StreamWrapper: wrapped_stderr: int = 0 wrapped_excepthook: int = 0 capturing: int = 0 - listeners: set + listeners: set[_ProgressListener] def __init__(self) -> None: self.stdout = self.original_stdout = sys.stdout @@ -277,14 +305,14 @@ def __init__(self) -> None: if env.env_flag('WRAP_STDERR', default=False): # pragma: no cover self.wrap_stderr() - def start_capturing(self, bar: ProgressBarMixinBase | None = None) -> None: + def start_capturing(self, bar: _ProgressListener | None = None) -> None: if bar: # pragma: no branch self.listeners.add(bar) self.capturing += 1 self.update_capturing() - def stop_capturing(self, bar: ProgressBarMixinBase | None = None) -> None: + def stop_capturing(self, bar: _ProgressListener | None = None) -> None: if bar: # pragma: no branch with contextlib.suppress(KeyError): self.listeners.remove(bar) @@ -403,7 +431,7 @@ def excepthook( self, exc_type: type[BaseException], exc_value: BaseException, - exc_traceback: types.TracebackType | None, + exc_traceback: TracebackType | None, ) -> None: self.original_excepthook(exc_type, exc_value, exc_traceback) self.flush() @@ -455,13 +483,13 @@ class AttributeDict(dict): AttributeError: No such attribute: spam """ - def __getattr__(self, name: str) -> int: + def __getattr__(self, name: str) -> typing.Any: if name in self: return self[name] else: raise AttributeError(f'No such attribute: {name}') - def __setattr__(self, name: str, value: int) -> None: + def __setattr__(self, name: str, value: typing.Any) -> None: self[name] = value def __delattr__(self, name: str) -> None: diff --git a/progressbar/widgets.py b/progressbar/widgets.py index 147f6cf..a8fd750 100644 --- a/progressbar/widgets.py +++ b/progressbar/widgets.py @@ -1,6 +1,7 @@ from __future__ import annotations import abc +import collections.abc import contextlib import datetime import functools @@ -11,12 +12,12 @@ # `types` module from typing import ClassVar -from python_utils import containers, converters, types +from python_utils import containers, converters from . import algorithms, base, terminal, utils from .terminal import colors -if types.TYPE_CHECKING: +if typing.TYPE_CHECKING: from .bar import NumberT, ProgressBarMixinBase logger = logging.getLogger(__name__) @@ -25,13 +26,15 @@ MAX_TIME = datetime.time.max MAX_DATETIME = datetime.datetime.max -Data = types.Dict[str, types.Any] -FormatString = typing.Optional[str] +Data = dict[str, typing.Any] +FormatString = str | None T = typing.TypeVar('T') -def string_or_lambda(input_): +def string_or_lambda( + input_: str | collections.abc.Callable[..., str], +) -> collections.abc.Callable[..., str]: if isinstance(input_, str): def render_input(progress, data, width): @@ -42,7 +45,9 @@ def render_input(progress, data, width): return input_ -def create_wrapper(wrapper): +def create_wrapper( + wrapper: str | tuple[str | None, str | None] | None, +) -> str | None: """Convert a wrapper tuple or format string to a format string. >>> create_wrapper('') @@ -87,32 +92,39 @@ def wrap(*args: typing.Any, **kwargs: typing.Any): return wrap -def create_marker(marker, wrap=None): - def _marker(progress, data, width): - if ( - progress.max_value is not base.UnknownLength - and progress.max_value > 0 - ): - # The fill length is based on the progress relative to - # min_value; the max() guards against a zero range and the - # min() keeps the marker within the allotted width when the - # value exceeds max_value (with max_error=False) - length = min( - width, - int( - (progress.value - progress.min_value) - / max(progress.max_value - progress.min_value, 1e-6) - * width, - ), - ) - return marker * length - else: - return marker - +def create_marker( + marker: str | collections.abc.Callable[..., str], + wrap: str | tuple[str | None, str | None] | None = None, +) -> collections.abc.Callable[..., str]: if isinstance(marker, str): - marker = converters.to_unicode(marker) - if utils.len_color(marker) != 1: + # Narrow to ``str`` once, in a fresh local, so the ``_marker`` closure + # below closes over a plain ``str`` (no cast needed). ``_marker`` is + # only ever wrapped in this branch. + marker_str = converters.to_unicode(marker) + if utils.len_color(marker_str) != 1: raise ValueError('Markers are required to be 1 char') + + def _marker(progress, data, width): + if ( + progress.max_value is not base.UnknownLength + and progress.max_value > 0 + ): + # The fill length is based on the progress relative to + # min_value; the max() guards against a zero range and the + # min() keeps the marker within the allotted width when the + # value exceeds max_value (with max_error=False) + length = min( + width, + int( + (progress.value - progress.min_value) + / max(progress.max_value - progress.min_value, 1e-6) + * width, + ), + ) + return marker_str * length + else: + return marker_str + return wrapper(_marker, wrap) else: return wrapper(marker, wrap) @@ -147,7 +159,7 @@ def get_format( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ) -> str: return format or self.format @@ -155,7 +167,7 @@ def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ) -> str: """Formats the widget into a string.""" format_ = self.get_format(progress, data, format) @@ -228,13 +240,13 @@ def check_size(self, progress: ProgressBarMixinBase): class TGradientColors(typing.TypedDict): - fg: types.Optional[terminal.OptionalColor | None] - bg: types.Optional[terminal.OptionalColor | None] + fg: terminal.OptionalColor + bg: terminal.OptionalColor class TFixedColors(typing.TypedDict): - fg_none: types.Optional[terminal.Color | None] - bg_none: types.Optional[terminal.Color | None] + fg_none: terminal.Color | None + bg_none: terminal.Color | None class WidgetBase(WidthWidgetMixin, metaclass=abc.ABCMeta): @@ -289,7 +301,7 @@ def __call__(self, progress: ProgressBarMixinBase, data: Data) -> str: # _fixed_colors: ClassVar[dict[str, terminal.Color | None]] = dict() # _gradient_colors: ClassVar[dict[str, terminal.OptionalColor | None]] = ( # dict()) - _len: typing.Callable[[str | bytes], int] = len + _len: collections.abc.Callable[[str | bytes], int] = len @functools.cached_property def uses_colors(self): @@ -386,7 +398,7 @@ class FormatLabel(FormatWidgetMixin, WidgetBase): """ - mapping: ClassVar[types.Dict[str, types.Tuple[str, types.Any]]] = dict( + mapping: ClassVar[dict[str, tuple[str, typing.Any]]] = dict( finished=('end_time', None), last_update=('last_update_time', None), max=('max_value', None), @@ -403,7 +415,7 @@ def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ): for name, (key, transform) in self.mapping.items(): # Avoid a per-entry contextlib.suppress on the redraw hot path: a @@ -470,7 +482,7 @@ class SamplesMixin(TimeSensitiveWidgetBase, metaclass=abc.ABCMeta): def __init__( self, - samples=datetime.timedelta(seconds=2), + samples: datetime.timedelta | int = datetime.timedelta(seconds=2), key_prefix=None, **kwargs, ): @@ -557,12 +569,13 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value, - elapsed, - ): + value: float, + elapsed: datetime.timedelta | None, + ) -> float: """Updates the widget to show the ETA or total time when finished.""" if elapsed: - # The max() prevents zero division errors + # The max() prevents zero division errors. ``value`` is always a + # number here (``_resolve_value_elapsed`` fills the default). per_item = elapsed.total_seconds() / max(value, 1e-6) remaining = progress.max_value - data['value'] return remaining * per_item @@ -649,9 +662,9 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value, - elapsed, - ): + value: float, + elapsed: datetime.timedelta | None, + ) -> datetime.datetime: eta_seconds = ETA._calculate_eta(self, progress, data, value, elapsed) now = datetime.datetime.now() try: @@ -780,7 +793,7 @@ def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ): value = data[self.variable] if value is not None: @@ -891,12 +904,12 @@ class AnimatedMarker(TimeSensitiveWidgetBase): def __init__( self, - markers='|/-\\', - default=None, - fill='', - marker_wrap=None, - fill_wrap=None, - **kwargs, + markers: str = '|/-\\', + default: str | None = None, + fill: str = '', + marker_wrap: str | tuple[str | None, str | None] | None = None, + fill_wrap: str | tuple[str | None, str | None] | None = None, + **kwargs: typing.Any, ): self.markers = markers self.marker_wrap = create_wrapper(marker_wrap) @@ -1011,10 +1024,10 @@ class SimpleProgress(FormatWidgetMixin, ColoredMixin, WidgetBase): max_width_cache: dict[ str | tuple[ - NumberT | types.Type[base.UnknownLength] | None, - NumberT | types.Type[base.UnknownLength] | None, + NumberT | type[base.UnknownLength] | None, + NumberT | type[base.UnknownLength] | None, ], - types.Optional[int], + int | None, ] DEFAULT_FORMAT = '%(value_s)s of %(max_value_s)s' @@ -1053,7 +1066,7 @@ def __call__( # Guess the maximum width from the min and max value key = progress.min_value, progress.max_value - max_width: types.Optional[int] = self.max_width_cache.get( + max_width: int | None = self.max_width_cache.get( key, self.max_width, ) @@ -1086,8 +1099,8 @@ def __call__( class Bar(AutoWidthWidgetBase): """A progress bar which stretches to fill the line.""" - fg: terminal.OptionalColor | None = colors.gradient - bg: terminal.OptionalColor | None = None + fg: terminal.OptionalColor = colors.gradient + bg: terminal.OptionalColor = None def __init__( self, @@ -1139,7 +1152,7 @@ def __call__( progress: ProgressBarMixinBase, data: Data, width: int = 0, - color=True, + color: bool = True, ): """Updates the progress bar and its subcomponents.""" left, right, width = self._render_borders(progress, data, width) @@ -1200,7 +1213,7 @@ def __call__( progress: ProgressBarMixinBase, data: Data, width: int = 0, - color=True, + color: bool = True, ): """Updates the progress bar and its subcomponents.""" left, right, width = self._render_borders(progress, data, width) @@ -1227,16 +1240,17 @@ def __call__( class FormatCustomText(FormatWidgetMixin, WidgetBase): - mapping: types.Dict[str, types.Any] = dict() # noqa: RUF012 + mapping: dict[str, typing.Any] = dict() # noqa: RUF012 copy = False def __init__( self, format: str, - mapping: types.Optional[types.Dict[str, types.Any]] = None, + mapping: dict[str, typing.Any] | None = None, **kwargs, ): - self.format = format + # ``self.format`` is set by FormatWidgetMixin.__init__ (via super + # below); assigning it here too would just overwrite that. # Always build a fresh per-instance dict so update_mapping() never # mutates the shared class-level default (which every other # default-constructed instance would otherwise alias). Fall back to @@ -1245,14 +1259,14 @@ def __init__( self.mapping = dict(self.mapping if mapping is None else mapping) super().__init__(format=format, **kwargs) - def update_mapping(self, **mapping: types.Any): + def update_mapping(self, **mapping: typing.Any): self.mapping.update(mapping) def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ): return FormatWidgetMixin.__call__( self, @@ -1298,7 +1312,7 @@ def __call__( progress: ProgressBarMixinBase, data: Data, width: int = 0, - color=True, + color: bool = True, ): """Updates the progress bar and its subcomponents.""" left, right, width = self._render_borders(progress, data, width) @@ -1309,7 +1323,7 @@ def __call__( middle = '' values_accumulated = 0 width_accumulated = 0 - for marker, value in zip(self.markers, values): + for marker, value in zip(self.markers, values, strict=False): marker = converters.to_unicode(marker(progress, data, width)) if progress.custom_len(marker) != 1: raise ValueError('Markers are required to be 1 char') @@ -1540,7 +1554,7 @@ def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ): value = data['variables'][self.name] context = data.copy() @@ -1588,7 +1602,7 @@ def __call__( self, progress: ProgressBarMixinBase, data: Data, - format: types.Optional[str] = None, + format: str | None = None, ): data['current_time'] = self.current_time() data['current_datetime'] = self.current_datetime() @@ -1659,7 +1673,6 @@ def __init__( failure_marker='X', **kwargs, ): - self.name = name # Retained for backward compatibility only; render state now lives in # ``progress.extra`` (see get_job_markers) so a single widget reused by # multiple bars no longer interleaves their markers. @@ -1667,9 +1680,6 @@ def __init__( # Unique per-widget key so multiple JobStatusBars on the same bar do # not share storage either. self._markers_key = f'{type(self).__name__}_{id(self)}_job_markers' - self.left = string_or_lambda(left) - self.right = string_or_lambda(right) - self.fill = string_or_lambda(fill) self.success_fg_color = success_fg_color self.success_bg_color = success_bg_color self.success_marker = success_marker @@ -1698,7 +1708,7 @@ def __call__( progress: ProgressBarMixinBase, data: Data, width: int = 0, - color=True, + color: bool = True, ): left, right, width = self._render_borders(progress, data, width) diff --git a/ruff.toml b/ruff.toml index c36fefd..a105697 100644 --- a/ruff.toml +++ b/ruff.toml @@ -1,7 +1,7 @@ # We keep the ruff configuration separate so it can easily be shared across # all projects -target-version = 'py39' +target-version = 'py310' #src = ['progressbar'] exclude = [ diff --git a/tests/conftest.py b/tests/conftest.py index d8f0ea2..bad96e6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,10 @@ from __future__ import annotations +import collections.abc import logging import time import timeit +import typing from datetime import datetime, timezone import freezegun @@ -18,14 +20,14 @@ } -def pytest_configure(config) -> None: +def pytest_configure(config: pytest.Config) -> None: logging.basicConfig( level=LOG_LEVELS.get(config.option.verbose, logging.DEBUG), ) @pytest.fixture(autouse=True) -def disable_native_accelerator(monkeypatch): +def disable_native_accelerator(monkeypatch: pytest.MonkeyPatch) -> None: # The optional native accelerator (speedups.FastBarIterator) is exercised # explicitly in test_native_accelerator.py. Every other test targets the # pure-Python iterator (`_iter_python`), so force that path by default when @@ -37,7 +39,9 @@ def disable_native_accelerator(monkeypatch): @pytest.fixture(autouse=True) -def small_interval(monkeypatch, request) -> None: +def small_interval( + monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest +) -> None: # Tests marked `no_freezegun` need real timing conditions (e.g. the perf # budget test), so preserve the default _MINIMUM_UPDATE_INTERVAL so the # fast-path gate can calibrate and activate correctly. @@ -53,7 +57,9 @@ def small_interval(monkeypatch, request) -> None: @pytest.fixture(autouse=True) -def sleep_faster(monkeypatch, request): +def sleep_faster( + monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest +) -> collections.abc.Iterator[typing.Any]: # Tests marked `no_freezegun` need a real, advancing clock (e.g. the # gate's perf test, which only activates after a real timing measurement). # For those, skip the freezegun wrapping entirely. diff --git a/tests/test_init_exports.py b/tests/test_init_exports.py new file mode 100644 index 0000000..0f09096 --- /dev/null +++ b/tests/test_init_exports.py @@ -0,0 +1,63 @@ +"""Guard the three hand-synced export lists in ``progressbar/__init__.py``. + +``_NAME_TO_MODULE`` is the single source of truth for the lazily re-exported +public names. ``__all__`` and the ``TYPE_CHECKING`` import block must stay in +sync with it; these tests fail loudly if any of the three drift apart. +""" + +from __future__ import annotations + +import ast +import pathlib + +import progressbar + +# Alias (not a `from` import) so CodeQL doesn't flag `progressbar` as imported +# with both `import` and `import from`. +_NAME_TO_MODULE = progressbar._NAME_TO_MODULE + +#: Dunders that are eagerly imported (not part of ``_NAME_TO_MODULE``) but are +#: still part of the public ``__all__``. +_EAGER_DUNDERS: frozenset[str] = frozenset({'__author__', '__version__'}) + + +def test_every_mapping_name_resolves() -> None: + # Each lazily re-exported name must be reachable via attribute access + # (which drives ``__getattr__``'s import machinery). + for name in _NAME_TO_MODULE: + assert getattr(progressbar, name) is not None, name + + +def test_all_matches_mapping_plus_dunders() -> None: + # ``__all__`` must contain exactly the mapping names plus the eager + # dunders. The concrete ordering is delegated to ruff's RUF022, so this + # compares contents rather than the exact list order. + assert set(progressbar.__all__) == set(_NAME_TO_MODULE) | _EAGER_DUNDERS + + +def test_all_has_no_duplicates() -> None: + assert len(progressbar.__all__) == len(set(progressbar.__all__)) + + +def test_type_checking_block_imports_exactly_the_mapping() -> None: + source: str = pathlib.Path(progressbar.__file__).read_text() + tree: ast.Module = ast.parse(source) + + imported: set[str] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.If): + continue + + test = node.test + is_type_checking = ( + isinstance(test, ast.Attribute) and test.attr == 'TYPE_CHECKING' + ) or (isinstance(test, ast.Name) and test.id == 'TYPE_CHECKING') + if not is_type_checking: + continue + + for stmt in node.body: + if isinstance(stmt, ast.ImportFrom): + for alias in stmt.names: + imported.add(alias.asname or alias.name) + + assert imported == set(_NAME_TO_MODULE) diff --git a/tests/test_subclass_compat.py b/tests/test_subclass_compat.py index c4910e2..01c9152 100644 --- a/tests/test_subclass_compat.py +++ b/tests/test_subclass_compat.py @@ -81,11 +81,10 @@ def __call__(self, progress, data, format=None): class OldStyleSamplesWidget(widgets.SamplesMixin): def __init__(self, **kwargs: typing.Any): - # samples accepts int per the class docstring/doctest; the - # timedelta-only annotation is a known typing gap (PR 4). + # samples accepts int per the class docstring/doctest. widgets.SamplesMixin.__init__( self, - samples=3, # pyright: ignore[reportArgumentType] + samples=3, **kwargs, )