From 36bb4002bc9652974a362dfdeed18941f40d8c93 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:26:58 +0200 Subject: [PATCH 01/18] refactor(types): PEP 604/585 sweep in bar.py Replace python_utils.types typing aliases with modern spellings: Optional[X] -> X | None, Union -> |, Dict -> dict, Type -> type, Iterable/Iterator/Sequence/MutableSequence/Callable -> collections.abc.*, TypeVar/cast/Any -> typing.*. Drop the now-unused python_utils.types import. Annotation-only; no runtime semantics change. --- progressbar/bar.py | 83 +++++++++++++++++++++++----------------------- 1 file changed, 42 insertions(+), 41 deletions(-) diff --git a/progressbar/bar.py b/progressbar/bar.py index ab996f7..89946c6 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 = '' @@ -337,14 +338,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 +357,12 @@ 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'))) + self.fd.write(typing.cast(str, line.encode('ascii', 'replace'))) 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 +565,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 +662,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 +694,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 +722,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 +752,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 +768,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. @@ -800,7 +801,7 @@ def _setup_poll_intervals( ) # type: ignore def _seed_variables( - self, variables: types.Optional[types.Dict[str, typing.Any]] + self, variables: dict[str, typing.Any] | None ) -> None: """Seed the user-defined variables dict and scan widgets for names. @@ -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: From a520ba28e19a9fdd6eb20fa67ce5865a8721606d Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:33:52 +0200 Subject: [PATCH 02/18] refactor(types): PEP 604/585 sweep in multi/utils/fast/shortcuts Modernize typing aliases: Optional[X] -> X | None, Set -> set, Type -> type, and Iterable/Iterator/Sequence/Callable/Generator -> collections.abc.*. Route TYPE_CHECKING/TypeVar through typing; keep python_utils.types only for its own StringTypes alias in utils.py. Annotation-only; no runtime semantics change. --- progressbar/fast.py | 3 ++- progressbar/multi.py | 10 ++++++---- progressbar/shortcuts.py | 8 +++++--- progressbar/utils.py | 17 +++++++++-------- 4 files changed, 22 insertions(+), 16 deletions(-) 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..eaca368 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, @@ -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 diff --git a/progressbar/shortcuts.py b/progressbar/shortcuts.py index c4d2c5d..af1947a 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.Iterator[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/utils.py b/progressbar/utils.py index 3cfc14f..035b18a 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,7 +19,7 @@ from progressbar import base, env, terminal -if types.TYPE_CHECKING: +if typing.TYPE_CHECKING: from .bar import ProgressBar, ProgressBarMixinBase # Make sure these are available for import @@ -28,7 +29,7 @@ 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) # 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 @@ -140,7 +141,7 @@ def __init__( self, target: base.IO, capturing: bool = False, - listeners: types.Optional[types.Set[ProgressBar]] = None, + listeners: set[ProgressBar] | None = None, ) -> None: self.buffer = io.StringIO() self.target = target @@ -214,7 +215,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 +248,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, ], @@ -403,7 +404,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() From 6a27bca887bf12c16aaecacaf784f298682a37a0 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:35:13 +0200 Subject: [PATCH 03/18] refactor(types): PEP 604/585 sweep in terminal/* Modernize typing aliases in terminal/base.py and terminal/stream.py: Optional[X] -> X | None, Union -> |, List/Tuple -> list/tuple, Callable/Generator -> collections.abc.*, cast/Any -> typing.*. Drop the now-unused python_utils.types import from base.py. Annotation-only. --- progressbar/terminal/base.py | 35 +++++++++++++++++----------------- progressbar/terminal/stream.py | 4 ++-- 2 files changed, 20 insertions(+), 19 deletions(-) diff --git a/progressbar/terminal/base.py b/progressbar/terminal/base.py index 8a79fbc..7ddc32a 100644 --- a/progressbar/terminal/base.py +++ b/progressbar/terminal/base.py @@ -2,6 +2,7 @@ import abc import collections +import collections.abc import colorsys import enum import threading @@ -12,7 +13,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 +170,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 +202,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. @@ -401,7 +402,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,19 +449,19 @@ def __hash__(self) -> int: class Colors: - by_name: ClassVar[defaultdict[str, types.List[Color]]] = ( + by_name: ClassVar[defaultdict[str, list[Color]]] = ( collections.defaultdict(list) ) - by_lowername: ClassVar[defaultdict[str, types.List[Color]]] = ( + by_lowername: ClassVar[defaultdict[str, list[Color]]] = ( collections.defaultdict(list) ) - by_hex: ClassVar[defaultdict[str, types.List[Color]]] = ( + by_hex: ClassVar[defaultdict[str, list[Color]]] = ( collections.defaultdict(list) ) - by_rgb: ClassVar[defaultdict[RGB, types.List[Color]]] = ( + by_rgb: ClassVar[defaultdict[RGB, list[Color]]] = ( collections.defaultdict(list) ) - by_hls: ClassVar[defaultdict[HSL, types.List[Color]]] = ( + by_hls: ClassVar[defaultdict[HSL, list[Color]]] = ( collections.defaultdict(list) ) by_xterm: ClassVar[dict[int, Color]] = dict() @@ -469,9 +470,9 @@ class Colors: 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 +498,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 +556,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 +573,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: From f6638974dc4f23ba1b313c43ba5c96909747b90c Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:36:54 +0200 Subject: [PATCH 04/18] refactor(types): PEP 604/585 sweep in widgets.py Modernize typing aliases: Optional[X] -> X | None, Dict/Tuple -> dict/ tuple, Type -> type, Callable -> collections.abc.Callable, Any -> typing.Any, and route TYPE_CHECKING through typing. Collapse redundant Optional[X | None] color-dict fields to X | None. Drop the now-unused python_utils.types import. Annotation-only; no runtime semantics change. --- progressbar/widgets.py | 49 +++++++++++++++++++++--------------------- 1 file changed, 25 insertions(+), 24 deletions(-) diff --git a/progressbar/widgets.py b/progressbar/widgets.py index 147f6cf..d97b09c 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,8 +26,8 @@ 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') @@ -147,7 +148,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 +156,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 +229,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 | None + bg: terminal.OptionalColor | None 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 +290,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 +387,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 +404,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 @@ -780,7 +781,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: @@ -1011,10 +1012,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 +1054,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, ) @@ -1227,13 +1228,13 @@ 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 @@ -1245,14 +1246,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, @@ -1540,7 +1541,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 +1589,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() From 987dbdccdd01e1a3fd30dd26b3d764e60f2bfeb6 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:52:50 +0200 Subject: [PATCH 05/18] refactor(types): add missing hot-path and public annotations widgets.py: annotate string_or_lambda/create_marker/create_wrapper, _calculate_eta (ETA/AbsoluteETA), AnimatedMarker.__init__, Bar-family __call__ color: bool, and SamplesMixin.samples (timedelta | int, per its docstring). multi.py: return annotations on __setitem__/__getitem__/stop/ get_sorted_bars/__enter__. utils.py: AttributeDict __getattr__/__setattr__ -> Any, and typed WrappingIO/StreamWrapper listeners as set[ProgressBarMixinBase] (matches start_capturing's parameter type). conftest.py: fixture annotations. Drop the now-obsolete SamplesMixin pyright ignore in tests. Annotation-only. --- progressbar/multi.py | 10 +++--- progressbar/utils.py | 12 +++---- progressbar/widgets.py | 63 ++++++++++++++++++++++------------- tests/conftest.py | 14 +++++--- tests/test_subclass_compat.py | 5 ++- 5 files changed, 62 insertions(+), 42 deletions(-) diff --git a/progressbar/multi.py b/progressbar/multi.py index eaca368..8f1a0c2 100644 --- a/progressbar/multi.py +++ b/progressbar/multi.py @@ -163,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 @@ -188,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) @@ -413,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/utils.py b/progressbar/utils.py index 035b18a..421c55b 100644 --- a/progressbar/utils.py +++ b/progressbar/utils.py @@ -20,7 +20,7 @@ from progressbar import base, env, terminal if typing.TYPE_CHECKING: - from .bar import ProgressBar, ProgressBarMixinBase + from .bar import ProgressBarMixinBase # Make sure these are available for import assert timedelta_to_seconds is not None @@ -134,14 +134,14 @@ class WrappingIO: buffer: io.StringIO target: base.IO capturing: bool - listeners: set + listeners: set[ProgressBarMixinBase] needs_clear: bool = False def __init__( self, target: base.IO, capturing: bool = False, - listeners: set[ProgressBar] | None = None, + listeners: set[ProgressBarMixinBase] | None = None, ) -> None: self.buffer = io.StringIO() self.target = target @@ -260,7 +260,7 @@ class StreamWrapper: wrapped_stderr: int = 0 wrapped_excepthook: int = 0 capturing: int = 0 - listeners: set + listeners: set[ProgressBarMixinBase] def __init__(self) -> None: self.stdout = self.original_stdout = sys.stdout @@ -456,13 +456,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 d97b09c..4f97661 100644 --- a/progressbar/widgets.py +++ b/progressbar/widgets.py @@ -32,7 +32,9 @@ 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): @@ -43,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('') @@ -88,8 +92,15 @@ def wrap(*args: typing.Any, **kwargs: typing.Any): return wrap -def create_marker(marker, wrap=None): +def create_marker( + marker: str | collections.abc.Callable[..., str], + wrap: str | tuple[str | None, str | None] | None = None, +) -> collections.abc.Callable[..., str]: def _marker(progress, data, width): + # ``marker`` is narrowed to ``str`` by the caller below (``_marker`` is + # only wrapped when ``marker`` is a string); cast keeps the type + # checker honest without a runtime effect. + marker_str = typing.cast(str, marker) if ( progress.max_value is not base.UnknownLength and progress.max_value > 0 @@ -106,9 +117,9 @@ def _marker(progress, data, width): * width, ), ) - return marker * length + return marker_str * length else: - return marker + return marker_str if isinstance(marker, str): marker = converters.to_unicode(marker) @@ -471,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, ): @@ -558,13 +569,17 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value, - elapsed, - ): + value: float | None, + 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 - per_item = elapsed.total_seconds() / max(value, 1e-6) + # The max() prevents zero division errors. ``value`` is resolved + # to a number by ``_resolve_value_elapsed`` before it reaches + # here; the cast documents that without changing runtime behavior. + per_item = elapsed.total_seconds() / max( + typing.cast(float, value), 1e-6 + ) remaining = progress.max_value - data['value'] return remaining * per_item else: @@ -650,9 +665,9 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value, - elapsed, - ): + value: float | None, + elapsed: datetime.timedelta | None, + ) -> datetime.datetime: eta_seconds = ETA._calculate_eta(self, progress, data, value, elapsed) now = datetime.datetime.now() try: @@ -892,12 +907,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) @@ -1140,7 +1155,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) @@ -1201,7 +1216,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) @@ -1299,7 +1314,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) @@ -1699,7 +1714,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/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_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, ) From 7c897d02e451060fd39f8db8730737c4373467ce Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:53:57 +0200 Subject: [PATCH 06/18] refactor(types): widen progressbar() iterator param to Iterable The shortcut only needs an iterable (ProgressBar.__call__ iterates it), so accept collections.abc.Iterable[T] instead of Iterator[T]. This lets progressbar(range(10)) type-check. Parameter name 'iterator' kept for keyword-caller compatibility. Annotation-only. --- progressbar/shortcuts.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/progressbar/shortcuts.py b/progressbar/shortcuts.py index af1947a..df72eba 100644 --- a/progressbar/shortcuts.py +++ b/progressbar/shortcuts.py @@ -16,7 +16,7 @@ def progressbar( - iterator: collections.abc.Iterator[T], + iterator: collections.abc.Iterable[T], min_value: bar.NumberT = 0, max_value: bar.ValueT = None, widgets: collections.abc.Sequence[widgets_module.WidgetBase | str] From f8a51a33a8ef8f55a00f1383994385394580eb3c Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 05:57:41 +0200 Subject: [PATCH 07/18] refactor(types): overload deltas_to_seconds sentinel default The old signature (default: Optional[Type[ValueError]]) was a typing lie (callers pass floats/None as the default). Add @overloads: with the ValueError sentinel it returns float (or raises); with any other default T it returns float | T. Implementation signature stays permissive (Any), runtime behavior unchanged. --- progressbar/utils.py | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/progressbar/utils.py b/progressbar/utils.py index 421c55b..53fe13e 100644 --- a/progressbar/utils.py +++ b/progressbar/utils.py @@ -30,6 +30,7 @@ assert epoch is not None StringT = typing.TypeVar('StringT', bound=types.StringTypes) +T = typing.TypeVar('T') # 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 @@ -40,10 +41,24 @@ ) +@typing.overload def deltas_to_seconds( *deltas: None | datetime.timedelta | float, - default: types.Optional[types.Type[ValueError]] = ValueError, -) -> int | float | None: + default: type[ValueError] = ..., +) -> float: ... + + +@typing.overload +def deltas_to_seconds( + *deltas: None | datetime.timedelta | float, + default: T, +) -> float | T: ... + + +def deltas_to_seconds( + *deltas: None | datetime.timedelta | float, + default: typing.Any = ValueError, +) -> typing.Any: """ Convert timedeltas and seconds as int to seconds as float while coalescing. @@ -81,8 +96,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: From 9bbcfd1fad2ac871a54b4f36078bc8b5eec74115 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:00:03 +0200 Subject: [PATCH 08/18] style(types): bump ruff target-version to py310 Bump target-version 'py39' -> 'py310'. The only newly-flagged rule is B905 (zip without strict=); add explicit strict=False at the two call sites to preserve the stop-at-shortest behavior (no new exception path). No pyupgrade (UP) autofixes were triggered. --- progressbar/terminal/base.py | 20 +++++++++++--------- progressbar/widgets.py | 2 +- ruff.toml | 2 +- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/progressbar/terminal/base.py b/progressbar/terminal/base.py index 7ddc32a..5e75d9e 100644 --- a/progressbar/terminal/base.py +++ b/progressbar/terminal/base.py @@ -226,7 +226,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, @@ -449,20 +451,20 @@ def __hash__(self) -> int: class Colors: - by_name: ClassVar[defaultdict[str, list[Color]]] = ( - collections.defaultdict(list) + by_name: ClassVar[defaultdict[str, list[Color]]] = collections.defaultdict( + list ) by_lowername: ClassVar[defaultdict[str, list[Color]]] = ( collections.defaultdict(list) ) - by_hex: ClassVar[defaultdict[str, list[Color]]] = ( - collections.defaultdict(list) + by_hex: ClassVar[defaultdict[str, list[Color]]] = collections.defaultdict( + list ) - by_rgb: ClassVar[defaultdict[RGB, list[Color]]] = ( - collections.defaultdict(list) + by_rgb: ClassVar[defaultdict[RGB, list[Color]]] = collections.defaultdict( + list ) - by_hls: ClassVar[defaultdict[HSL, list[Color]]] = ( - collections.defaultdict(list) + by_hls: ClassVar[defaultdict[HSL, list[Color]]] = collections.defaultdict( + list ) by_xterm: ClassVar[dict[int, Color]] = dict() diff --git a/progressbar/widgets.py b/progressbar/widgets.py index 4f97661..5746619 100644 --- a/progressbar/widgets.py +++ b/progressbar/widgets.py @@ -1325,7 +1325,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') 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 = [ From 51cc68d6c9c1f6c7cbc43416684590ed4475cf6e Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:04:42 +0200 Subject: [PATCH 09/18] test(init): guard __init__ export lists against drift Add tests/test_init_exports.py asserting the three hand-synced lists stay consistent: every _NAME_TO_MODULE name resolves via getattr, __all__'s contents equal the mapping plus the eager dunders, and the TYPE_CHECKING import block (parsed via ast) imports exactly the mapping's names. Reorder __all__ into its canonical form and document _NAME_TO_MODULE as the source of truth. __all__ stays a static literal (kept in RUF022 order) so type checkers still see the re-exports. --- progressbar/__init__.py | 4 +++ tests/test_init_exports.py | 60 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) create mode 100644 tests/test_init_exports.py 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/tests/test_init_exports.py b/tests/test_init_exports.py new file mode 100644 index 0000000..8fde904 --- /dev/null +++ b/tests/test_init_exports.py @@ -0,0 +1,60 @@ +"""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 +from progressbar import _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) From ee8c2ecf4454f6532a2e87a34a42425a55c40a33 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:43:43 +0200 Subject: [PATCH 10/18] fix(utils): break bar<->utils import cycle; harden deltas_to_seconds typing - Replace the TYPE_CHECKING `from .bar import ProgressBarMixinBase` and its listener annotations with a local `_ProgressListener` Protocol, so utils has no dependency on bar (CodeQL flagged the type-checking-only edge as a module-level cyclic import; no runtime cycle existed). Clears both the utils.py and bar.py cyclic-import alerts by removing the only utils->bar edge. - Give the two deltas_to_seconds @overload stubs docstring bodies instead of `...` (CodeQL 'statement has no effect'). - Add `int` to the *deltas parameter type to match the documented/implemented behavior. Annotation-only; runtime unchanged. --- progressbar/utils.py | 39 ++++++++++++++++++++++++++------------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/progressbar/utils.py b/progressbar/utils.py index 53fe13e..ae7e612 100644 --- a/progressbar/utils.py +++ b/progressbar/utils.py @@ -19,9 +19,6 @@ from progressbar import base, env, terminal -if typing.TYPE_CHECKING: - from .bar import ProgressBarMixinBase - # Make sure these are available for import assert timedelta_to_seconds is not None assert get_terminal_size is not None @@ -32,6 +29,20 @@ 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 # every redraw. @@ -43,20 +54,22 @@ @typing.overload def deltas_to_seconds( - *deltas: None | datetime.timedelta | float, + *deltas: None | datetime.timedelta | float | int, default: type[ValueError] = ..., -) -> float: ... +) -> float: + """Coalesce to seconds; raise ``ValueError`` if no delta is valid.""" @typing.overload def deltas_to_seconds( - *deltas: None | datetime.timedelta | float, + *deltas: None | datetime.timedelta | float | int, default: T, -) -> float | T: ... +) -> float | T: + """Coalesce to seconds; return ``default`` if no delta is valid.""" def deltas_to_seconds( - *deltas: None | datetime.timedelta | float, + *deltas: None | datetime.timedelta | float | int, default: typing.Any = ValueError, ) -> typing.Any: """ @@ -148,14 +161,14 @@ class WrappingIO: buffer: io.StringIO target: base.IO capturing: bool - listeners: set[ProgressBarMixinBase] + listeners: set[_ProgressListener] needs_clear: bool = False def __init__( self, target: base.IO, capturing: bool = False, - listeners: set[ProgressBarMixinBase] | None = None, + listeners: set[_ProgressListener] | None = None, ) -> None: self.buffer = io.StringIO() self.target = target @@ -274,7 +287,7 @@ class StreamWrapper: wrapped_stderr: int = 0 wrapped_excepthook: int = 0 capturing: int = 0 - listeners: set[ProgressBarMixinBase] + listeners: set[_ProgressListener] def __init__(self) -> None: self.stdout = self.original_stdout = sys.stdout @@ -292,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) From 479f7f462364861ea592c7042edce8e08b84ff26 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:44:58 +0200 Subject: [PATCH 11/18] fix(bar): write ASCII-safe str in DefaultFdMixin.update fallback The UnicodeEncodeError fallback encoded to bytes and wrote them to the text stream fd (masked by a typing.cast(str, ...)), which raises TypeError at runtime. Decode the ASCII-encoded bytes back to str before writing. Fixes a latent bug in the (uncovered) fallback path. --- progressbar/bar.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/progressbar/bar.py b/progressbar/bar.py index 89946c6..4405432 100644 --- a/progressbar/bar.py +++ b/progressbar/bar.py @@ -357,7 +357,10 @@ def update(self, *args: typing.Any, **kwargs: typing.Any) -> None: try: # pragma: no cover self.fd.write(line) except UnicodeEncodeError: # pragma: no cover - self.fd.write(typing.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, @@ -800,9 +803,7 @@ def _setup_poll_intervals( float(os.environ.get('PROGRESSBAR_MINIMUM_UPDATE_INTERVAL', 0)), ) # type: ignore - def _seed_variables( - self, variables: dict[str, typing.Any] | None - ) -> 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`` From 34e9e1524316e56482068d08eda1ee557a62d03d Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:47:48 +0200 Subject: [PATCH 12/18] refactor(widgets): simplify marker/ETA typing per review - create_marker: define _marker inside the isinstance(str) branch, closing over a fresh str local, so the typing.cast(str, marker) is no longer needed (behavior-identical; _marker was only ever wrapped in that branch). - _calculate_eta (ETA + AbsoluteETA): tighten value to float (callers pass a value resolved by _resolve_value_elapsed), dropping the typing.cast(float). - Drop redundant '| None' from terminal.OptionalColor fields (OptionalColor already includes None). --- progressbar/widgets.py | 75 ++++++++++++++++++++---------------------- 1 file changed, 36 insertions(+), 39 deletions(-) diff --git a/progressbar/widgets.py b/progressbar/widgets.py index 5746619..4c67e4d 100644 --- a/progressbar/widgets.py +++ b/progressbar/widgets.py @@ -96,35 +96,35 @@ def create_marker( marker: str | collections.abc.Callable[..., str], wrap: str | tuple[str | None, str | None] | None = None, ) -> collections.abc.Callable[..., str]: - def _marker(progress, data, width): - # ``marker`` is narrowed to ``str`` by the caller below (``_marker`` is - # only wrapped when ``marker`` is a string); cast keeps the type - # checker honest without a runtime effect. - marker_str = typing.cast(str, marker) - 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 - 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) @@ -240,8 +240,8 @@ def check_size(self, progress: ProgressBarMixinBase): class TGradientColors(typing.TypedDict): - fg: terminal.OptionalColor | None - bg: terminal.OptionalColor | None + fg: terminal.OptionalColor + bg: terminal.OptionalColor class TFixedColors(typing.TypedDict): @@ -569,17 +569,14 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value: float | None, + 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. ``value`` is resolved - # to a number by ``_resolve_value_elapsed`` before it reaches - # here; the cast documents that without changing runtime behavior. - per_item = elapsed.total_seconds() / max( - typing.cast(float, value), 1e-6 - ) + # 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 else: @@ -665,7 +662,7 @@ def _calculate_eta( self, progress: ProgressBarMixinBase, data: Data, - value: float | None, + value: float, elapsed: datetime.timedelta | None, ) -> datetime.datetime: eta_seconds = ETA._calculate_eta(self, progress, data, value, elapsed) @@ -1102,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, From c5ead2f61a25dbafaf97748584eb80b110ee9b4b Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:48:30 +0200 Subject: [PATCH 13/18] test(init): use import-alias for progressbar in export test Reference progressbar._NAME_TO_MODULE via an alias instead of mixing `import progressbar` with `from progressbar import ...` (CodeQL py/import-and-import-from), matching tests/test_fast_default.py. --- tests/test_init_exports.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_init_exports.py b/tests/test_init_exports.py index 8fde904..0f09096 100644 --- a/tests/test_init_exports.py +++ b/tests/test_init_exports.py @@ -11,7 +11,10 @@ import pathlib import progressbar -from progressbar import _NAME_TO_MODULE + +# 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__``. From a970484e38d4b77026582d6ffbebc15aa28f1750 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 06:52:40 +0200 Subject: [PATCH 14/18] fix(bar): cooperatively chain __del__ to superclass finalizer ProgressBarMixinBase.__del__ ran finish() but never chained to a super __del__ (CodeQL 'missing call to superclass __del__', re-surfaced after the ProgressBarBase base-class change). Add a hasattr-guarded super().__del__() call: a no-op today (abc.ABC/object define no __del__) but keeps finalization cooperative and avoids an AttributeError if a base with __del__ is ever added. --- progressbar/bar.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/progressbar/bar.py b/progressbar/bar.py index 4405432..819760d 100644 --- a/progressbar/bar.py +++ b/progressbar/bar.py @@ -175,6 +175,13 @@ def __del__(self): except Exception: # noqa: BLE001, S110 pass + # Cooperatively chain to a superclass finalizer when one exists. This + # class's bases (abc.ABC, object) define no __del__, so the call is a + # no-op today; the guard keeps finalization cooperative — and avoids + # an AttributeError — should a base with __del__ ever be introduced. + if hasattr(super(), '__del__'): # pragma: no cover + super().__del__() # pyright: ignore[reportAttributeAccessIssue] + def __getstate__(self): return self.__dict__ From 1395a470ac5bb8b95f64add8736485bfb6e04e20 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 07:36:45 +0200 Subject: [PATCH 15/18] style(terminal): drop plain 'import collections' (CodeQL import-and-import-from) Remove the plain 'import collections' that CodeQL flagged as mixing with 'from collections import defaultdict'. Keep the from-import (defaultdict is a public re-export recorded in the API snapshot) and reference defaultdict bare; 'import collections.abc' stays for collections.abc.Callable. --- progressbar/terminal/base.py | 21 +++++---------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/progressbar/terminal/base.py b/progressbar/terminal/base.py index 5e75d9e..88eb51d 100644 --- a/progressbar/terminal/base.py +++ b/progressbar/terminal/base.py @@ -1,7 +1,6 @@ from __future__ import annotations import abc -import collections import collections.abc import colorsys import enum @@ -451,21 +450,11 @@ def __hash__(self) -> int: class Colors: - by_name: ClassVar[defaultdict[str, list[Color]]] = collections.defaultdict( - list - ) - by_lowername: ClassVar[defaultdict[str, list[Color]]] = ( - collections.defaultdict(list) - ) - by_hex: ClassVar[defaultdict[str, list[Color]]] = collections.defaultdict( - list - ) - by_rgb: ClassVar[defaultdict[RGB, list[Color]]] = collections.defaultdict( - list - ) - by_hls: ClassVar[defaultdict[HSL, 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 From 151e151d080ef4c55b98b565127fa96fc7c01a6c Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 07:36:45 +0200 Subject: [PATCH 16/18] fix(bar): give ProgressBarBase an explicit __del__ chain (CodeQL missing-call-to-delete) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeQL flagged ProgressBarBase (Iterable + ProgressBarMixinBase multiple inheritance) for finalizing without an explicit __del__ calling the base finalizer. Revert the earlier wrong-class guarded super().__del__() in ProgressBarMixinBase and define ProgressBarBase.__del__ = super().__del__(), which resolves to the same ProgressBarMixinBase.__del__ reached by inheritance — behaviourally identical, MI chain now explicit. --- progressbar/bar.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/progressbar/bar.py b/progressbar/bar.py index 819760d..4b88f3f 100644 --- a/progressbar/bar.py +++ b/progressbar/bar.py @@ -175,13 +175,6 @@ def __del__(self): except Exception: # noqa: BLE001, S110 pass - # Cooperatively chain to a superclass finalizer when one exists. This - # class's bases (abc.ABC, object) define no __del__, so the call is a - # no-op today; the guard keeps finalization cooperative — and avoids - # an AttributeError — should a base with __del__ ever be introduced. - if hasattr(super(), '__del__'): # pragma: no cover - super().__del__() # pyright: ignore[reportAttributeAccessIssue] - def __getstate__(self): return self.__dict__ @@ -214,6 +207,15 @@ def __repr__(self): label = f': {self.label}' if self.label else '' return f'<{self.__class__.__name__}#{self.index}{label}>' + def __del__(self) -> None: + # ProgressBarBase mixes collections.abc.Iterable in alongside + # ProgressBarMixinBase (which defines the real finalizer). Define + # __del__ here so that finalizer is reached explicitly through the MRO + # instead of only by attribute inheritance — behaviourally identical + # (super() resolves to the same ProgressBarMixinBase.__del__), but it + # makes the multiple-inheritance finalizer chain unambiguous. + super().__del__() # pragma: no cover + class DefaultFdMixin(ProgressBarMixinBase): # The file descriptor to write to. Defaults to `sys.stderr` From b950458a78ce14dfc293d452f912e950f0b0c781 Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 07:38:50 +0200 Subject: [PATCH 17/18] fix(widgets): drop redundant self.* assignments (CodeQL overwritten-inherited-attribute) FormatCustomText.__init__ and JobStatusBar.__init__ set instance attributes (format; name/left/right/fill) that their super().__init__ immediately re-sets from the same values, which CodeQL flags as overwriting the inherited attribute. Remove the duplicates; the base initializers (FormatWidgetMixin, Bar, VariableMixin) set them. Verified attrs identical after construction; no behavior change. --- progressbar/widgets.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/progressbar/widgets.py b/progressbar/widgets.py index 4c67e4d..a8fd750 100644 --- a/progressbar/widgets.py +++ b/progressbar/widgets.py @@ -1249,7 +1249,8 @@ def __init__( 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 @@ -1672,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. @@ -1680,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 From eff9faf4bf72ceb4c4cd166bd8e3a1ce2fdf0d6f Mon Sep 17 00:00:00 2001 From: Rick van Hattem Date: Mon, 6 Jul 2026 08:03:14 +0200 Subject: [PATCH 18/18] revert(bar): drop finalizer-chain workarounds, keep develop __del__ semantics Two attempts to satisfy CodeQL's py/missing-call-to-delete approximation added finalizer complexity with zero behavioral gain: ProgressBarBase defines no __del__ of its own and runtime attribute lookup reaches ProgressBarMixinBase.__del__ unconditionally; CodeQL's static super() model simply cannot follow the MRO past the Iterable hop. Restore the exact develop semantics (teardown-fragile code should stay simple) and dismiss the alert as a false positive instead. --- progressbar/bar.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/progressbar/bar.py b/progressbar/bar.py index 4b88f3f..ed35a72 100644 --- a/progressbar/bar.py +++ b/progressbar/bar.py @@ -207,16 +207,6 @@ def __repr__(self): label = f': {self.label}' if self.label else '' return f'<{self.__class__.__name__}#{self.index}{label}>' - def __del__(self) -> None: - # ProgressBarBase mixes collections.abc.Iterable in alongside - # ProgressBarMixinBase (which defines the real finalizer). Define - # __del__ here so that finalizer is reached explicitly through the MRO - # instead of only by attribute inheritance — behaviourally identical - # (super() resolves to the same ProgressBarMixinBase.__del__), but it - # makes the multiple-inheritance finalizer chain unambiguous. - super().__del__() # pragma: no cover - - class DefaultFdMixin(ProgressBarMixinBase): # The file descriptor to write to. Defaults to `sys.stderr` fd: base.TextIO = sys.stderr