Quality audit PR 4/6: typing modernization

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',

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,44 +64,44 @@
 # 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.
     term_width: int = 80
     #: 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,35 +116,35 @@
     #: 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
     #: unless the `max_error` parameter is `False`.
     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 __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 @@
         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 @@
         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 @@
 
         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 @@
         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 @@
         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 @@
     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 @@
             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 _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 @@
         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 _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 @@
             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 @@
 
         return percentage
 
-    def data(self) -> types.Dict[str, types.Any]:
+    def data(self) -> dict[str, typing.Any]:
         """
 
         Returns:

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

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