# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""An indeterminate terminal spinner for long-running, blocking work.
Click ships {func}`click.progressbar`, but it is *determinate*: it needs a known
length or an iterable to advance through. Some work has no measurable progress:
a blocking subprocess, a network round-trip, a query whose duration is unknown.
For those, the only honest feedback is "something is happening".
{class}`Spinner` fills that gap. It animates a small frame sequence on a daemon
thread, so the caller can stay blocked in a single call (`communicate()`,
`urlopen()`, ...) while the spinner keeps turning:
```{code-block} python
from time import sleep
from click_extra import Spinner
with Spinner("Brewing tea"):
sleep(5) # A blocking call with no measurable progress.
```
```{caution}
The spinner draws with carriage returns and ANSI control codes, so it is a
no-op whenever its output stream is not a TTY (a pipe, a file, a captured
test buffer, a CI log), unless `live="always"` forces it. This keeps
redirected output and machine-readable formats clean.
```
```{note}
On Windows, {meth}`Spinner.start` enables the console's virtual-terminal
processing so the ANSI control codes animate in place rather than print
literally (`β β[0m β¦ β[K`). Modern terminals (Windows Terminal, recent
conhost) already have it on; this just covers older consoles.
```
"""
from __future__ import annotations
import functools
import os
import signal
import sys
import threading
import time
from gettext import gettext as _
from typing import Final, TypeVar, cast
import click
from click._utils import UNSET
from extra_platforms import is_windows
from . import context
from ._deprecated import warn_deprecated_argument
from .color import (
COLOR_DISABLING_TERMS,
invocation_color,
is_a_tty,
resolve_color_env,
)
from .humanize import format_duration
from .layout import cell_width
from .parameters import ExtraOption
from .spinner_presets import (
SPINNER_FRAMES,
SPINNERS,
SpinnerPreset,
)
from .styling import Style
from .theme import KO_GLYPH, OK_GLYPH, get_current_theme
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Iterable, Sequence
from types import FrameType, TracebackType
from typing import IO, Any, Literal, Protocol, TextIO, TypeAlias
from click._termui_impl import ProgressBar
from click._utils import T_UNSET
from typing_extensions import Self
TLive: TypeAlias = Literal["auto", "always", "never"]
"""Where a spinner or a progress bar may draw, see {data}`LIVE_MODES`."""
class _LiveLine(Protocol):
"""A live terminal line other output must cooperate with.
The shared surface {func}`_active_line` exposes so a concurrent writer
(the logging bridge, a trail) can print above whichever live line owns
the terminal right now, be it a {class}`Spinner` or an
{class}`OperationTrail`'s progress-bar indicator.
"""
def _resolve_stream(self) -> IO[str]: ...
def echo(self, message: str) -> None: ...
class _AggregateIndicator(Protocol):
"""The live aggregate indicator an {class}`OperationTrail` drives.
Both the spinner-backed and bar-backed indicators expose this surface,
so the trail drives either one the same way: enter it, {meth}`advance`
the tally as outcomes land, {meth}`echo` a persistent line per outcome,
then {meth}`finish` with a kept summary.
"""
@property
def shown(self) -> bool: ...
def __enter__(self) -> Self: ...
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None: ...
def advance(self, done: int) -> None: ...
def echo(self, message: str) -> None: ...
def finish(self, ok: bool, summary: str) -> None: ...
def stop(self) -> None: ...
_ACTIVE_LINES: list[_LiveLine] = []
"""Stack of the live terminal lines currently drawing, innermost last.
A {class}`Spinner` or an {class}`OperationTrail` progress-bar indicator
registers itself when it actually begins drawing (a disabled one never
registers) and deregisters when it stops. Guarded by
{data}`_ACTIVE_LINES_LOCK`, since these are started and stopped from worker
threads too.
"""
_ACTIVE_LINES_LOCK = threading.Lock()
"""Guards {data}`_ACTIVE_LINES` against concurrent mutation."""
def _restore_cursor_on_sigterm(signum: int, frame: FrameType | None) -> None:
"""Show the cursor again on every live line's terminal, then die by `SIGTERM`.
A live line hides the cursor while it draws, and `SIGTERM` kills the process
before any clean-up runs, which leaves the user's terminal with no cursor
unless this handler shows it first. The registry is read without its lock:
the code this handler interrupted may hold it, and a list copy is atomic.
"""
for line in _ACTIVE_LINES.copy():
try:
stream = line._resolve_stream()
stream.write("\r\x1b[K\x1b[?25h")
stream.flush()
except (OSError, ValueError):
# A closed or broken stream cannot show the cursor anymore.
pass
signal.signal(signal.SIGTERM, signal.SIG_DFL)
os.kill(os.getpid(), signal.SIGTERM)
def _guard_cursor(installing: bool) -> None:
"""Install or remove {func}`_restore_cursor_on_sigterm` for live lines.
Only from the main thread, the one {func}`signal.signal` accepts, and only
over the default action: a program handling `SIGTERM` keeps its handler.
Windows has no `SIGTERM` worth guarding: termination there runs no handler.
A line stopped from another thread leaves the handler in place, which then
finds no live line and only re-delivers the signal.
"""
if is_windows() or threading.current_thread() is not threading.main_thread():
return
current = signal.getsignal(signal.SIGTERM)
if installing and current is signal.SIG_DFL:
signal.signal(signal.SIGTERM, _restore_cursor_on_sigterm)
elif not installing and current is _restore_cursor_on_sigterm:
signal.signal(signal.SIGTERM, signal.SIG_DFL)
def _register_line(line: _LiveLine) -> None:
"""Advertise `line` as a live terminal line. Idempotent."""
with _ACTIVE_LINES_LOCK:
if line not in _ACTIVE_LINES:
_ACTIVE_LINES.append(line)
_guard_cursor(installing=True)
def _deregister_line(line: _LiveLine) -> None:
"""Withdraw `line` from the live-line registry. Idempotent."""
with _ACTIVE_LINES_LOCK:
if line in _ACTIVE_LINES:
_ACTIVE_LINES.remove(line)
empty = not _ACTIVE_LINES
if empty:
_guard_cursor(installing=False)
def _active_line(stream: IO[str] | None = None) -> _LiveLine | None:
"""Return the innermost live terminal line drawing, or `None`.
With `stream` given, only a line drawing on that very stream matches. This
is how output producers cooperate with a running animation or progress bar
instead of garbling it: {class}`click_extra.logging.StreamHandler` checks
here and routes its records through the line's `echo`, which erases the
in-progress render, prints the line, and redraws underneath.
"""
with _ACTIVE_LINES_LOCK:
for line in reversed(_ACTIVE_LINES):
if stream is None or line._resolve_stream() is stream:
return line
return None
[docs]
def active_spinner(stream: IO[str] | None = None) -> Spinner | None:
"""Return the innermost {class}`Spinner` currently animating, or `None`.
A spinner-typed view of {func}`_active_line`, skipping any progress-bar
indicator that may own the line instead. With `stream` given, only a
spinner drawing on that very stream matches.
"""
with _ACTIVE_LINES_LOCK:
for line in reversed(_ACTIVE_LINES):
if isinstance(line, Spinner) and (
stream is None or line._resolve_stream() is stream
):
return line
return None
def _color_enabled(stream: IO[str]) -> bool:
"""Decide whether output to `stream` may carry ANSI color.
Follows the invocation's reconciled color first, so `--color` / `--no-color`
and the `NO_COLOR` / `FORCE_COLOR` family have already been honored. It is
read through {func}`~click_extra.color.invocation_color`, never `ctx.color`
alone, because a trail line and a spinner can be rendered on a worker thread
the thread-local command context does not reach. Outside a CLI it falls back
to the color environment variables and a dumb/unknown `TERM`, read by
{func}`~click_extra.color.resolve_color_env`, then to TTY detection.
"""
color = invocation_color()
if color is not None:
return color
env_color = resolve_color_env()
if env_color is not None:
return env_color
return is_a_tty(stream)
LIVE_MODES: tuple[str, ...] = ("auto", "always", "never")
"""The values of the `live` argument of {class}`Spinner` and {class}`OperationTrail`.
- `"auto"`: draw only on an interactive terminal that can move the cursor, and
only when the `--progress` flag of the active command allows it.
- `"always"`: draw on any stream, a pipe or a captured buffer included.
- `"never"`: never draw.
"""
def _check_live(live: str) -> None:
"""Reject a `live` value outside {data}`LIVE_MODES`.
:raises ValueError: if `live` is not one of {data}`LIVE_MODES`.
"""
if live not in LIVE_MODES:
raise ValueError('live must be "auto", "always" or "never".')
def _progress_enabled() -> bool:
"""Whether the active command's `--progress` flag allows a live display.
Follows {data}`~click_extra.context.PROGRESS`, which `--no-progress` and
`--accessible` turn off. `True` outside a command context, and on a thread the
thread-local context does not reach.
"""
ctx = click.get_current_context(silent=True)
return ctx is None or bool(context.get(ctx, context.PROGRESS, True))
def _can_draw(live: str, stream: IO[str]) -> bool:
"""Resolve whether a cursor-driven display may draw on `stream`.
`"always"` and `"never"` decide on their own. `"auto"` draws only when the
active command's `--progress` flag allows it (see {func}`_progress_enabled`),
and only on an interactive terminal that can move the cursor. That rules out
non-interactive streams (a pipe, file or captured buffer, which are not a
TTY) and `TERM=dumb` / `TERM=unknown` terminals, whose lack of cursor
control would smear the output instead of updating it in place. Shared by
{class}`Spinner` and the {class}`OperationTrail` progress-bar indicator.
"""
if live != "auto":
return live == "always"
if not _progress_enabled():
return False
if os.environ.get("TERM", "").lower() in COLOR_DISABLING_TERMS:
return False
return is_a_tty(stream)
def _stream_or_stderr(stream: IO[str] | None) -> IO[str]:
"""The stream a line is drawn on: the one given, or {data}`sys.stderr`.
Read at the moment of drawing rather than at construction, so a stream
swapped in afterwards (as test harnesses do) is honored.
"""
return stream if stream is not None else sys.stderr
_STOP_GRACE: Final = 1.0
"""Seconds a stopping display waits for its drawing thread before leaving it behind.
A drawing thread finishes its frame within milliseconds, unless the stream stops
draining: a terminal paused by flow control (Ctrl+S), or a full pipe. Waiting on
it without a bound would then hang the stop, and with it the Ctrl+C that asked
for the stop. The thread is a daemon, so it is left behind, and the clean-up
writes are skipped: they would block on the same stream.
"""
[docs]
class Spinner:
"""A thread-animated, indeterminate progress spinner usable as a context
manager.
The animation runs on a background daemon thread, leaving the calling thread
free to block on the actual work. Entering the context (or calling
{meth}`start`) begins the animation; leaving it (or calling {meth}`stop`)
halts the thread and erases the spinner line so it never lingers above the
next output.
```{note}
A single {class}`Spinner` instance drives one animation at a time. mpm
and similar tools run their subprocesses sequentially, so one shared
instance whose {attr}`label` is reassigned between steps is enough; for
concurrent work, use one instance per thread.
```
"""
label: str
"""Text drawn after the spinner glyph.
Reassign it at any time while the spinner runs to reflect the current step;
the animation thread reads it afresh on every frame.
"""
def __init__(
self,
label: str | Callable[..., Any] = "",
*,
frames: Sequence[str] | None = None,
spinner: SpinnerPreset | None = None,
reverse: bool = False,
interval: float | None = None,
delay: float = 0.0,
style: Style | None = None,
label_style: Style | None = None,
timer: bool | Callable[[float], str] | None = None,
timer_style: Style | None = None,
stream: IO[str] | None = None,
live: TLive = "auto",
hide_cursor: bool = True,
beep: bool = False,
enabled: bool | T_UNSET | None = UNSET,
) -> None:
"""Configure (but do not start) the spinner.
:param label: text shown after the spinner glyph. As a special case, a
bare `@Spinner` decorator passes the wrapped function here instead;
it is detected and the label defaults to empty.
:param frames: the animation frames, cycled in order. Defaults to
{data}`~click_extra.spinner_presets.SPINNER_FRAMES`, or the `spinner`
preset's frames when given.
:param spinner: a {class}`~click_extra.spinner_presets.SpinnerPreset` from
the {data}`~click_extra.spinner_presets.SPINNERS` catalog
(`spinner=SPINNERS["moon"]`), supplying both frames and a tuned
interval. An explicit `frames` or `interval` still overrides it.
:param reverse: cycle the frames backwards, spinning the animation the
other way. Set it when the rotation runs counter to what you expect;
it composes with any custom `frames`.
:param interval: seconds between two frames. Defaults to `0.1`, or the
`spinner` preset's interval when given.
:param delay: seconds to wait before drawing the first frame. A non-zero
delay keeps the spinner silent for calls that finish quickly, so it
only surfaces once an operation is genuinely slow.
:param style: a {class}`~click_extra.styling.Style` applied to the spinner
glyph, label and timer (`Style(fg="cyan", bold=True)`). Color is
decoupled from animation: `--no-color` / `NO_COLOR` strip it while
the spinner keeps spinning (see {class}`ProgressOption`). The strip
reaches escapes embedded in `label` itself too, so a label styled
by hand is as safe as one styled through these arguments.
:param label_style: a {class}`~click_extra.styling.Style` for the label
alone, in place of `style` there. It also paints the label of the
kept {meth}`ok` / {meth}`fail` line, which `style` leaves plain. It
outlives a reassigned {attr}`label`, where styling embedded in the
text has to be applied again at every step.
:param timer: append the elapsed wall-clock time to the spinner, and to
any final {meth}`ok` / {meth}`fail` line. `None` (the default)
follows the CLI's `--time` / `--no-time` flag, as an
{class}`OperationTrail` does. `True` forces it on with
{func}`~click_extra.humanize.format_duration`'s compact format
(`2.3s`, `1:05`, then `1:02:03`), `False` forces it off, and a
callable `(seconds: float) -> str` formats the duration yourself,
like ``timer=lambda s: f"{s / 60:.0f}m"`` for whole minutes.
:param timer_style: a {class}`~click_extra.styling.Style` for the timer,
parentheses included, in place of `style` there
(`Style(dim=True)` to set the clock back from the label). A `timer`
callable formats only the duration, so it cannot reach the
parentheses: this can. Also paints the timer of the kept {meth}`ok`
/ {meth}`fail` line.
:param stream: where to draw; defaults to {data}`sys.stderr` so the
spinner never mixes into `stdout` data.
:param live: where the animation may draw: `"auto"` (the default) on an
interactive terminal only, unless `--no-progress` turns it off,
`"always"` on any stream, `"never"` nowhere. A spinner that does not
draw still writes its {meth}`ok` / {meth}`fail` line.
:param hide_cursor: hide the text cursor while spinning and restore it on
stop.
:param beep: ring the terminal bell once when the spinner stops. It
fires only when the spinner was active, so a disabled or redirected
spinner stays silent.
:param enabled: deprecated, use `live` instead: `None` stands for
`"auto"`, `True` for `"always"` and `False` for `"never"`.
:raises ValueError: if `style`, `label_style` or `timer_style` carries a
color or attribute that cannot be rendered, or if `live` is not one
of {data}`~click_extra.spinner.LIVE_MODES`.
"""
# Support a bare `@Spinner` decorator (no parentheses): the first
# positional is then the wrapped function, not a text label. `@Spinner(β¦)`
# and `with Spinner(β¦)` keep passing a string label as usual. A string is
# never callable, so this never misfires on a real label.
#
# This is the same `callable(first_arg)` test as
# `click_extra.decorators.allow_missing_parenthesis`, inlined here on
# purpose: that helper wraps a decorator *factory function* and returns a
# function, so it cannot wrap `Spinner` without replacing the class: and
# `Spinner` must stay a class to double as a context manager and to support
# `isinstance()` / subclassing. The bare-call hook therefore has to live
# in `__init__`, the one place the parenthesis-less form reaches.
self._decorated: Callable[..., Any] | None = None
if callable(label):
self._decorated = label
# Make the instance masquerade as the function it stands in for,
# without overwriting our own attributes (`updated=()`).
functools.update_wrapper(self, label, updated=())
label = ""
self.label = label
# `spinner=` supplies frames and interval together; an explicit `frames=`
# or `interval=` overrides the preset, and both fall back to the defaults.
if frames is not None:
self.frames = frames
elif spinner is not None:
self.frames = spinner.frames
else:
self.frames = SPINNER_FRAMES
if interval is not None:
self.interval = interval
elif spinner is not None:
self.interval = spinner.interval
else:
self.interval = 0.1
_check_live(live)
if enabled is not UNSET:
warn_deprecated_argument("Spinner", "enabled", "live=")
live = "auto" if enabled is None else "always" if enabled else "never"
self.reverse = reverse
self.delay = delay
self.style = style
self.label_style = label_style
# None follows the --time flag, resolved here on the constructing thread:
# the animation thread never sees the Click context the flag lives in.
self.timer = _resolve_timer(timer)
self.timer_style = timer_style
self.stream = stream
self.live = live
self.hide_cursor = hide_cursor
self.beep = beep
# Validate each style once, so a bad color or attribute fails loudly here
# instead of silently killing the draw thread (cloup builds and applies
# the style lazily on first call, where the error would surface off-thread).
for candidate in (style, label_style, timer_style):
if candidate is None:
continue
try:
candidate("")
except (TypeError, ValueError) as error:
raise ValueError(f"Invalid spinner style: {error}") from error
self._stop = threading.Event()
self._thread: threading.Thread | None = None
self._lock = threading.Lock()
self._drawn = False
self._cursor_hidden = False
self._color_enabled = False
self._start_time: float | None = None
self._stop_time: float | None = None
def _resolve_stream(self) -> IO[str]:
"""The `stream` given, or {data}`sys.stderr`: see {func}`_stream_or_stderr`."""
return _stream_or_stderr(self.stream)
def _resolve_live(self, stream: IO[str]) -> bool:
"""Decide whether to animate on `stream`, as {attr}`live` says.
See {func}`_can_draw` for the resolution.
"""
return _can_draw(self.live, stream)
def _resolve_color_enabled(self, stream: IO[str]) -> bool:
"""Decide whether to apply ANSI color, orthogonally to whether it animates.
See {func}`_color_enabled` for the resolution. This is independent of
{meth}`_resolve_live`: a spinner can spin in plain text (a TTY under
`NO_COLOR`), which is exactly the decoupling {class}`ProgressOption`
documents.
"""
return _color_enabled(stream)
def _style(self, text: str, *, color: bool | None = None) -> str:
"""Apply the configured {class}`~click_extra.styling.Style`, or return bare.
A no-op when no style was set or color is disabled, so the same call site
produces colored output on a capable terminal and plain output under
`NO_COLOR` / a pipe.
:param text: what to style.
:param color: override whether any color is applied. `None` follows what
{meth}`start` resolved for the spinner's own stream, which is what
the animation wants. A picture of the spinner overrides it, having
its own answer to whether ANSI survives.
:return: the text, styled or bare.
"""
return self._paint(text, self.style, color=color)
def _paint(
self, text: str, style: Style | None, *, color: bool | None = None
) -> str:
"""Apply `style` to `text` when color is on and there is text to paint.
:param style: the style to apply, used as given: no fallback to
{attr}`style`.
:param color: see {meth}`_style`.
:return: the text, styled or bare.
"""
enabled = self._color_enabled if color is None else color
if enabled and style is not None and text:
return style(text)
return text
@property
def elapsed_time(self) -> float:
"""Seconds elapsed since {meth}`start`, frozen once {meth}`stop` is called.
Returns `0.0` before the spinner has started.
"""
if self._start_time is None:
return 0.0
end = self._stop_time if self._stop_time is not None else time.monotonic()
return end - self._start_time
@property
def shown(self) -> bool:
"""Whether the spinner has drawn at least one frame to its stream.
`True` only once an animation frame was actually rendered. It stays
`False` for a disabled spinner (off a TTY, on a `TERM=dumb` terminal,
or with `live="never"`) and for a call that finishes within `delay`,
before the first frame. Reset by {meth}`start`.
Use it to gate output that should mirror the spinner's visibility.
{meth}`ok` and {meth}`fail` write their line unconditionally, so an
outcome is still recorded in a pipe or log; guard them with `shown` when
you only want the finisher on screen after a spinner the user actually
saw::
with Spinner("Baking bread") as spinner:
bake()
if spinner.shown:
spinner.ok()
"""
return self._drawn
def _clock(self) -> str:
"""The `( elapsed )` timer suffix, or empty when no timer is set.
`timer=True` uses {func}`~click_extra.humanize.format_duration`; a
callable `timer` formats {attr}`elapsed_time` itself. The result is
always wrapped the same way.
"""
if not self.timer:
return ""
return f" ({_format_timer(self.timer, self.elapsed_time)})"
@property
def _ordered_frames(self) -> tuple[str, ...]:
"""The frames in the order the animation cycles them, `reverse` applied."""
return tuple(reversed(self.frames)) if self.reverse else tuple(self.frames)
def _compose_frame(self, frame: str, *, color: bool | None = None) -> str:
"""Build the line one animation frame draws.
The glyph, the label and the timer, in the order and the styling the
animation writes them. Held in one place so a picture of a spinner shows
what the spinner draws, instead of a second guess at it that drifts the
first time this composition changes.
Without `label_style` or `timer_style`, `style` paints the line as one
run. With either, each part is painted on its own, carrying its leading
space, and a part with no style of its own falls back to `style`. With
color off, every escape is stripped, the ones embedded in the label
included.
:param frame: one of {attr}`frames`, the glyph the line opens with.
:param color: see {meth}`_style`.
:return: the line, ANSI escape sequences included when color is on.
"""
label = f" {self.label}" if self.label else ""
clock = self._clock()
if self.label_style is None and self.timer_style is None:
line = self._style(f"{frame}{label}{clock}", color=color)
else:
line = (
self._style(frame, color=color)
+ self._paint(
label,
self.style if self.label_style is None else self.label_style,
color=color,
)
+ self._paint(
clock,
self.style if self.timer_style is None else self.timer_style,
color=color,
)
)
enabled = self._color_enabled if color is None else color
return line if enabled else click.unstyle(line)
[docs]
def frame_lines(self, *, color: bool = True) -> tuple[str, ...]:
"""Every line this spinner's animation draws, one per frame.
One turn of the animation, held still. `reverse`, the label, the style
and the timer all land the way {meth}`start` would draw them, which is
what an animated capture stacks into a picture of this spinner.
```{note}
The timer is read once, here, so every line carries the same elapsed
time rather than a counting one: a still cannot show a clock running.
A spinner that never started reads zero.
```
:param color: style each line. On by default, because a capture renders
ANSI whatever the terminal the spinner would have drawn on accepts.
:return: the lines, in the order the animation cycles them.
"""
return tuple(
self._compose_frame(frame, color=color) for frame in self._ordered_frames
)
@staticmethod
def _enable_windows_ansi(stream: IO[str]) -> None:
"""Best-effort: turn on virtual-terminal processing for a Windows console.
Without it, legacy Windows consoles print the spinner's ANSI control codes
literally (`β β[0m β¦ β[K`) instead of animating in place: the recurring
complaint behind yaspin's Windows issues. Modern terminals (Windows
Terminal, recent conhost) already enable it; this just covers the
laggards. A no-op everywhere but Windows, and silent when the console (or
a non-console stream) refuses the mode.
"""
# Positive `sys.platform` guard so type checkers treat the body as
# platform-conditional rather than dead code on a non-Windows host.
if sys.platform == "win32":
try:
# Windows-only standard-library modules, imported lazily so the
# spinner module still loads on every platform.
import ctypes
import msvcrt
handle = msvcrt.get_osfhandle(stream.fileno())
kernel32 = ctypes.windll.kernel32
mode = ctypes.c_uint32()
if kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
enable_vt = 0x0004 # ENABLE_VIRTUAL_TERMINAL_PROCESSING
kernel32.SetConsoleMode(handle, mode.value | enable_vt)
except (OSError, ValueError, AttributeError):
# Raised for a non-console stream (no/closed fileno) or a console
# that refuses the mode; nothing actionable. On a modern terminal
# the codes already render, on a truly legacy one they cannot.
pass
[docs]
def start(self) -> None:
"""Begin animating on a background thread, unless the spinner is disabled.
A disabled spinner (non-TTY stream, or `live="never"`) returns at once
without spawning a thread or emitting anything (but still records the
start time, so a later {meth}`ok` / {meth}`fail` can report a duration).
"""
# Time the operation even when the spinner is silenced, and resolve color
# here on the calling thread: the animation thread never sees the Click
# context that `_resolve_color_enabled` reads.
self._start_time = time.monotonic()
self._stop_time = None
stream = self._resolve_stream()
self._color_enabled = self._resolve_color_enabled(stream)
if not self._resolve_live(stream):
return
# The spinner is about to emit ANSI control codes: make sure a Windows
# console will interpret rather than echo them.
self._enable_windows_ansi(stream)
self._stop.clear()
self._drawn = False
self._cursor_hidden = False
# Advertise the animation so concurrent writers (the logging bridge, see
# _active_line()) print through echo() instead of over the frame.
_register_line(self)
self._thread = threading.Thread(
target=self._animate,
args=(stream,),
daemon=True,
)
self._thread.start()
[docs]
def stop(self) -> None:
"""Halt the animation and erase the spinner line.
Idempotent and safe to call when the spinner never started. Restores the
cursor and clears the line only if the animation actually drew to the
terminal.
"""
# Freeze the timer first, before the early return, so even a never-drawn
# spinner reports the operation's duration through `elapsed_time`.
self._stop_time = time.monotonic()
# Withdraw from the active registry first, so a concurrent log record
# emitted during the teardown below goes through the plain path.
_deregister_line(self)
if self._thread is None:
return
self._stop.set()
self._thread.join(timeout=_STOP_GRACE)
stuck = self._thread.is_alive()
self._thread = None
if stuck:
# See _STOP_GRACE: the stream does not drain, so skip the clean-up.
return
# The animation thread has joined, so the draw lock is now free: take it
# so a concurrent `echo()` from another thread cannot interleave with the
# final cleanup. Joining before acquiring avoids deadlocking against the
# lock-holding frame write.
with self._lock:
# Undo only what was actually emitted: erase the line if a frame was
# drawn, and restore the cursor if it was hidden. Reaching this point
# means the spinner was active, so an opt-in bell rings here too: a
# disabled or redirected spinner returns above and stays silent.
cleanup = ""
if self._drawn:
cleanup += "\r\x1b[K"
if self._cursor_hidden:
cleanup += "\x1b[?25h"
if self.beep:
cleanup += "\a"
if cleanup:
stream = self._resolve_stream()
stream.write(cleanup)
stream.flush()
self._cursor_hidden = False
[docs]
def echo(self, message: str = "") -> None:
"""Print `message` on its own line above the running spinner.
Click's {func}`click.progressbar` and a bare `print` both fight the
animation: a frame drawn between the cursor returns and the text mangles
the line. {meth}`echo` takes the same draw lock as the animation thread,
erases the in-progress frame, writes `message` followed by a newline,
and lets the next tick redraw the spinner underneath. It is safe to call
from another thread while the spinner runs.
Output goes to the spinner's own `stream` (`stderr` by default), so
results written to `stdout` never need it. When the spinner is not
animating (disabled, or a non-TTY stream), it degrades to a plain write
of `message` with no control codes.
"""
stream = self._resolve_stream()
with self._lock:
# Erase the in-progress frame so the message starts at column 0.
if self._drawn:
stream.write("\r\x1b[K")
stream.write(f"{message}\n")
stream.flush()
[docs]
def ok(self, symbol: str | None = None, *, style: Style | None = None) -> None:
"""Stop the spinner and leave a persistent success line on screen.
Where {meth}`stop` erases the spinner, {meth}`ok` replaces the final
frame with `symbol` followed by the current label (and the elapsed time
when `timer` is set), then keeps that line. `symbol` defaults to the
themed success glyph {data}`~click_extra.theme.OK_GLYPH` (`β`), painted
with the active theme's `success` slot unless `style` overrides it.
Color is stripped under `--no-color` / `NO_COLOR`; the glyph stays.
"""
self._finalize(symbol, style, success=True)
[docs]
def fail(self, symbol: str | None = None, *, style: Style | None = None) -> None:
"""Stop the spinner and leave a persistent failure line on screen.
The failure counterpart of {meth}`ok`, defaulting to
{data}`~click_extra.theme.KO_GLYPH` (`β`) painted with the active
theme's `error` slot.
"""
self._finalize(symbol, style, success=False)
def _finalize(
self,
symbol: str | None,
style: Style | None,
*,
success: bool,
) -> None:
"""Stop the animation and write a kept ``{symbol} {label}`` final line.
Resolves color on the calling thread, stops the spinner (which erases the
live frame and restores the cursor), then writes the final line in its
place. The glyph and its paint default to the active theme's success /
error slots, so a finished spinner matches the rest of a themed CLI.
Degrades to a plain line when color is disabled or the spinner was never
shown, so the outcome is still recorded off a TTY.
"""
glyph = symbol if symbol is not None else (OK_GLYPH if success else KO_GLYPH)
if style is None:
theme = get_current_theme()
paint = theme.success if success else theme.error
else:
paint = style
stream = self._resolve_stream()
color_enabled = self._resolve_color_enabled(stream)
self.stop()
label = self._paint(
f" {self.label}" if self.label else "",
self.label_style,
color=color_enabled,
)
clock = self._paint(self._clock(), self.timer_style, color=color_enabled)
marker = paint(glyph) if color_enabled else glyph
line = f"{marker}{label}{clock}"
if not color_enabled:
line = click.unstyle(line)
with self._lock:
stream.write(f"{line}\n")
stream.flush()
def _animate(self, stream: IO[str]) -> None:
"""Frame loop run on the background thread.
Waits `delay` before the first frame, then writes one frame every
`interval` until {meth}`stop` is called. Every wait goes through the
stop {class}`~threading.Event`, so the spinner reacts to `stop()`
immediately instead of sleeping out the current interval. Stream errors
(a closed terminal) end the loop quietly rather than surfacing a
traceback from the background thread.
"""
# A call that finishes within `delay` never draws anything.
if self._stop.wait(self.delay):
return
# Resolve the rotation direction once: `reverse` flips the frame order.
frames = self._ordered_frames
try:
if self.hide_cursor:
stream.write("\x1b[?25l")
self._cursor_hidden = True
stream.flush()
index = 0
while not self._stop.is_set():
content = self._compose_frame(frames[index % len(frames)])
# Hold the draw lock so a concurrent `echo()` cannot interleave
# with a half-written frame. Return to the line start, then
# clear to end-of-line so a shrinking label leaves no stale
# characters behind.
with self._lock:
stream.write(f"\r{content}\x1b[K")
stream.flush()
self._drawn = True
index += 1
if self._stop.wait(self.interval):
break
except (OSError, ValueError):
# The stream was closed or detached mid-spin; nothing left to draw.
return
def __enter__(self) -> Self:
self.start()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
self.stop()
def __call__(self, *args: Any, **kwargs: Any) -> Any:
"""Use the spinner as a decorator, with or without parentheses.
`@Spinner` wraps a function directly; `@Spinner("Loading", β¦)` first
configures the spinner, then wraps. Either way the function spins for the
duration of every call and returns its result untouched. The one instance
is shared across calls, which is fine for sequential use; give concurrent
callers their own spinner.
"""
# Bare `@Spinner`: the instance stood in for the function (captured at
# construction), so calling it runs that function inside the context.
if self._decorated is not None:
with self:
return self._decorated(*args, **kwargs)
# `@Spinner(β¦)`: wrap the single function argument so each call spins.
(func,) = args
@functools.wraps(func)
def wrapper(*call_args: Any, **call_kwargs: Any) -> Any:
with self:
return func(*call_args, **call_kwargs)
return wrapper
[docs]
def trail_glyph(ok: bool) -> str:
"""Return the themed `β` or `β` glyph for a trail line or finisher.
The success glyph {data}`~click_extra.theme.OK_GLYPH` painted with the active
theme's `success` slot, or the failure glyph
{data}`~click_extra.theme.KO_GLYPH` painted with its `error` slot.
"""
theme = get_current_theme()
return theme.success(OK_GLYPH) if ok else theme.error(KO_GLYPH)
[docs]
def trail_line(ok: bool, message: str) -> str:
"""Format one `β`/`β` trail line: a status glyph followed by `message`."""
return f"{trail_glyph(ok)} {message}"
def _format_timer(timer: bool | Callable[[float], str], seconds: float) -> str:
"""Format `seconds` for a trail's timer suffix.
Uses {func}`~click_extra.humanize.format_duration` for `True` (the default
compact clock: `2.3s`, `1:05`, `1:02:03`), or the given callable for a
custom format. Callers guard on a truthy `timer` before calling.
"""
formatter = timer if callable(timer) else format_duration
return formatter(seconds)
def _time_flag_active() -> bool:
"""Whether the active command's `--time` flag is on.
`True` when a command context is active and carries the
{data}`~click_extra.context.START_TIME` marker that
{class}`~click_extra.execution.TimerOption` sets under `--time`; `False`
under `--no-time` (its default) or outside any command. The shared signal
behind `timer=None` on {class}`OperationTrail` and `show_eta=None` on
{func}`progressbar`, so a trail and a bare bar agree on when to show timing.
"""
ctx = click.get_current_context(silent=True)
return ctx is not None and context.get(ctx, context.START_TIME) is not None
def _resolve_timer(
timer: bool | Callable[[float], str] | None,
) -> bool | Callable[[float], str]:
"""Resolve a trail's `timer` setting, auto-detecting `--time` for `None`.
An explicit `bool` or callable is returned unchanged; `None` (the trail
default) follows the CLI's `--time` / `--no-time` flag via
{func}`_time_flag_active`, mirroring how `live="auto"` detects the terminal.
"""
return _time_flag_active() if timer is None else timer
class _SpinnerIndicator:
"""An {class}`OperationTrail` aggregate indicator backed by a {class}`Spinner`.
Carries the running ``{label} {done}/{total} {unit}`` tally on one animated
line while completed outcomes stream above it, and closes on the spinner's
kept {meth}`~Spinner.ok` / {meth}`~Spinner.fail` line. Used for a concurrent
batch, where per-call spinners would collide on the shared stream.
"""
def __init__(
self,
*,
label: str,
unit: str,
total: int,
delay: float,
live: TLive,
stream: IO[str] | None,
spinner: SpinnerPreset | None = None,
timer: bool | Callable[[float], str] = True,
clock: Literal["elapsed", "eta"] = "elapsed",
) -> None:
self._label = label
self._unit = unit
self._total = total
self._timer = timer
# In eta mode a hidden Click bar supplies the rolling-average estimate,
# shown in the spinner's label; the spinner's own elapsed timer is then
# off, and finish() appends the total elapsed itself (a done batch has no
# ETA). Reuses Click's make_step/format_eta rather than reimplementing it.
self._eta_bar: ProgressBar[int] | None = None
if timer and clock == "eta" and total > 0:
self._eta_bar = click.progressbar(range(total))
self._spinner = Spinner(
f"{label} 0/{total} {unit}",
spinner=spinner,
delay=delay,
live=live,
timer=False if self._eta_bar is not None else timer,
stream=stream,
)
@property
def shown(self) -> bool:
return self._spinner.shown
def __enter__(self) -> Self:
self._spinner.__enter__()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
self._spinner.__exit__(exc_type, exc_val, exc_tb)
def advance(self, done: int) -> None:
"""Re-label the spinner with the tally, adding the ETA in eta mode."""
label = f"{self._label} {done}/{self._total} {self._unit}"
if self._eta_bar is not None:
self._eta_bar.make_step(done - self._eta_bar.pos)
eta = self._eta_bar.format_eta() # "" until a step lets it estimate.
if eta:
label = f"{label} {eta}"
self._spinner.label = label
def echo(self, message: str) -> None:
self._spinner.echo(message)
def stop(self) -> None:
"""Halt the spinner and erase its line, with no kept line. Idempotent."""
self._spinner.stop()
def finish(self, ok: bool, summary: str) -> None:
"""Leave the spinner's kept `β`/`β` ``summary`` line, elapsed included."""
if not self._spinner.shown:
return
if self._eta_bar is not None:
# eta mode runs the spinner's own timer off, so append the batch's
# total elapsed here (a finished batch has no time remaining).
clock = _format_timer(self._timer, self._spinner.elapsed_time)
summary = f"{summary} ({clock})"
self._spinner.label = summary
(self._spinner.ok if ok else self._spinner.fail)()
# How often the bar refreshes to keep its running elapsed clock ticking. The bar
# has no animation thread of its own (unlike the spinner), so a daemon ticker
# redraws it at this cadence in `clock="elapsed"` mode; matches the spinner's
# default frame interval.
_BAR_TICK_INTERVAL = 0.1
class _BarIndicator:
"""An {class}`OperationTrail` aggregate indicator backed by a determinate
{func}`click.progressbar`.
Where {class}`_SpinnerIndicator` narrates an *indeterminate* pulse, this
carries a real ``{label} [####----] {done}/{total}`` bar (the trail knows
its `total`), with completed outcomes streaming above it exactly as they
do over a spinner. It drives Click's bar directly rather than iterating it:
{meth}`advance` steps and redraws it, {meth}`echo` erases it to slip a
persistent line above, then redraws it below.
```{note}
Click's {meth}`~click._termui_impl.ProgressBar.render_progress` skips a
redraw whose line is unchanged, so {meth}`_draw` clears the bar's
`_last_line` cache to force the post-`echo` redraw. Cursor hiding is left
to Click (its `BEFORE_BAR` / `AFTER_BAR`); this only restores the cursor
when it tears the bar down early.
```
"""
def __init__(
self,
*,
label: str,
unit: str,
total: int,
delay: float,
live: TLive,
stream: IO[str] | None,
timer: bool | Callable[[float], str] = True,
clock: Literal["elapsed", "eta"] = "elapsed",
) -> None:
self._label = label
self._unit = unit
self._total = total
self._delay = delay
self._live = live
self._stream = stream
self._timer = timer
self._clock = clock
self._lock = threading.Lock()
self._on = False
self._drawn = False
self._finished = False
self._start = 0.0
self._bar: ProgressBar[int] | None = None
self._stop_tick = threading.Event()
self._ticker: threading.Thread | None = None
def _resolve_stream(self) -> IO[str]:
return _stream_or_stderr(self._stream)
def __enter__(self) -> Self:
stream = self._resolve_stream()
self._on = _can_draw(self._live, stream)
self._start = time.monotonic()
# show_pos renders the `{done}/{total}` tally; item_show_func appends the
# counted unit after it (and, in elapsed mode, a running clock), echoing
# the spinner's `3/5 feeds` phrasing. The `range(total)` iterable is never
# consumed (advance() drives the bar directly); it just fixes the length
# and satisfies the typed overload that carries item_show_func.
self._bar = click.progressbar(
range(self._total),
label=self._label,
show_pos=True,
# Click's ETA (remaining time) is shown only in `clock="eta"` mode;
# `clock="elapsed"` renders a running elapsed clock through
# item_show_func instead, and a `timer`-off bar shows no time at all.
show_eta=bool(self._timer) and self._clock == "eta",
item_show_func=self._render_info,
file=cast("TextIO", stream),
hidden=not self._on,
)
if self._on:
# The bar emits ANSI control codes: make a legacy Windows console
# interpret rather than echo them, as Spinner.start does.
Spinner._enable_windows_ansi(stream)
_register_line(self)
# Draw the empty bar right away (unless a delay defers the first
# render), so its `0/total` state is visible from the start, like the
# spinner indicator's animated tally: without this the bar appears
# only when the first outcome advances it, leaving the screen blank
# while the batch is already running.
if self._delay <= 0:
with self._lock:
self._draw()
# An elapsed clock must keep ticking between outcomes; the bar has no
# animation thread, so drive periodic redraws from a daemon ticker.
if self._elapsed_clock():
self._ticker = threading.Thread(target=self._tick, daemon=True)
self._ticker.start()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
self.stop()
@property
def shown(self) -> bool:
return self._drawn
def _draw(self) -> None:
"""Force a bar redraw, defeating Click's unchanged-line dedup."""
assert self._bar is not None
self._bar._last_line = None
self._bar.render_progress()
self._drawn = True
def _elapsed_clock(self) -> bool:
"""Whether a running elapsed clock is drawn (timing on, `clock="elapsed"`)."""
return bool(self._timer) and self._clock == "elapsed"
def _render_info(self, item: int | None) -> str | None:
"""The `item_show_func`: a running elapsed clock plus the unit, or the unit.
In `clock="elapsed"` mode with timing on, the elapsed time since the bar
started is prepended to the counted unit (`5.0s feeds`); an ETA bar or a
timer-off bar shows just the unit (Click renders any ETA itself).
"""
if self._elapsed_clock():
clock = _format_timer(self._timer, time.monotonic() - self._start)
return f"{clock} {self._unit}" if self._unit else clock
return self._unit or None
def _tick(self) -> None:
"""Redraw periodically so the running elapsed clock keeps ticking.
The bar, unlike the spinner, has no animation thread; without this the
elapsed clock would freeze between outcomes. Runs only in elapsed mode:
an ETA needs no ticking, since Click recomputes it on each step.
"""
while not self._stop_tick.wait(_BAR_TICK_INTERVAL):
with self._lock:
if self._drawn and not self._finished:
self._draw()
def _stop_ticker(self) -> bool:
"""Stop and join the elapsed-clock ticker. Idempotent.
:return: `False` when the ticker is still blocked in a write after
{data}`_STOP_GRACE`, so the caller skips its own clean-up writes.
"""
if self._ticker is None:
return True
self._stop_tick.set()
self._ticker.join(timeout=_STOP_GRACE)
stuck = self._ticker.is_alive()
self._ticker = None
return not stuck
def advance(self, done: int) -> None:
"""Step the bar to `done` and redraw it, once past the initial delay."""
if not self._on:
return
assert self._bar is not None
with self._lock:
# A batch that finishes within `delay` never draws: remember the
# position and stay silent, as the spinner does for a quick batch.
if (time.monotonic() - self._start) < self._delay:
self._bar.pos = done
return
self._bar.make_step(done - self._bar.pos)
self._draw()
def echo(self, message: str) -> None:
"""Print `message` as a persistent line above the bar, then redraw it."""
stream = self._resolve_stream()
with self._lock:
if self._drawn:
stream.write("\r\x1b[K") # Erase the bar line.
stream.write(f"{message}\n") # Persistent line above.
if self._drawn:
self._draw() # Redraw the bar below.
stream.flush()
def finish(self, ok: bool, summary: str) -> None:
"""Replace the bar with a kept `β`/`β` ``summary`` line, elapsed included."""
_deregister_line(self)
if not self._stop_ticker():
self._finished = True
return
with self._lock:
self._finished = True
if not self._drawn:
return
stream = self._resolve_stream()
clock = (
f" ({_format_timer(self._timer, time.monotonic() - self._start)})"
if self._timer
else ""
)
line = f"{trail_line(ok, summary)}{clock}"
if not _color_enabled(stream):
line = click.unstyle(line)
# Erase the bar, keep the finisher in its place, restore the cursor
# Click hid via BEFORE_BAR.
stream.write(f"\r\x1b[K{line}\n\x1b[?25h")
stream.flush()
self._drawn = False
def stop(self) -> None:
"""Erase the bar and restore the cursor with no kept line. Idempotent.
The teardown path for an abnormal exit (an exception inside the trail's
`with` block, where {meth}`finish` never ran).
"""
_deregister_line(self)
if not self._stop_ticker():
self._finished = True
return
with self._lock:
if self._finished or not self._drawn:
self._finished = True
return
stream = self._resolve_stream()
stream.write("\r\x1b[K\x1b[?25h")
stream.flush()
self._drawn = False
self._finished = True
[docs]
class OperationTrail:
"""A `β`/`β` progress trail and finisher for a batch of operations.
Where {class}`Spinner` narrates *one* long-running call,
`OperationTrail` reports a *batch* of them: each completed operation
leaves a persistent {func}`~click_extra.spinner.trail_line` on screen, a
running `done/total` tally keeps the batch's pulse visible, and
{meth}`finish` closes with a persistent summary line. The natural
reporting companion of the concurrency primitives
{func}`~click_extra.execution.run_jobs` and
{func}`~click_extra.execution.run_lanes`, rendered one of three ways:
- **sequential** (`jobs <= 1`): echo each outcome as it lands, with no
aggregate indicator (each operation is free to keep its own per-call
{class}`Spinner`). {meth}`finish` appends the elapsed time.
- **concurrent** (`jobs > 1`): drive one aggregate {class}`Spinner`
(per-call spinners would collide on the shared stream), buffering
outcomes until it first draws, then streaming the rest live above it.
Pick the animation from the
{data}`~click_extra.spinner_presets.SPINNERS` catalog with `spinner=`.
- **progress bar** (`progress_bar=True`): drive one aggregate *determinate*
bar carrying the `{done}/{total}` tally, with outcomes streaming above
it. Serves sequential and concurrent batches alike, and needs a known
`total`.
The aggregate indicators redraw in place, which a pipe or a CI log cannot
do, so by default they draw only on an interactive terminal, and `live`
changes where they draw. The `β`/`β` lines and the finisher only append, so
they print on any stream, in plain text where color is off. Where no
indicator draws, every rendering echoes each outcome as it lands, as the
sequential one does. `visible=False` silences all of it. The running `β`
tally is kept as outcomes land ({attr}`ok_count`), so a caller computes no
counts of its own.
Thread-safe: {meth}`mark` may be called from worker threads. Use it as a
context manager whenever it may run concurrently, to bound the aggregate
spinner's life; a purely sequential caller may construct it bare.
```{code-block} python
from click_extra.execution import run_jobs
from click_extra.spinner import OperationTrail
with OperationTrail(label="Fetching", unit="feeds", total=len(feeds),
jobs=jobs) as trail:
def fetch(feed):
trail.mark(*pull(feed)) # pull() returns (ok, message).
list(run_jobs(fetch, feeds, jobs=jobs))
trail.finish(
trail.ok_count == len(feeds),
f"Fetched {trail.ok_count}/{len(feeds)} feeds",
)
```
"""
def __init__(
self,
*,
label: str = "",
unit: str = "",
total: int = 0,
jobs: int = 1,
spinner: SpinnerPreset | None = None,
progress_bar: bool = False,
timer: bool | Callable[[float], str] | None = None,
clock: Literal["elapsed", "eta"] = "elapsed",
visible: bool = True,
live: TLive = "auto",
echo_sequential: bool = True,
delay: float = 0.0,
stream: IO[str] | None = None,
enabled: bool | T_UNSET | None = UNSET,
) -> None:
"""Configure (but do not start) the trail.
:param label: present-tense verb for the running aggregate indicator
(`"Fetching"`), composed into its ``{label} {done}/{total} {unit}``
tally.
:param unit: the noun counted in the tally (`"files"`, `"feeds"`).
:param total: how many outcomes are expected, for the `done/total`
count.
:param jobs: the batch's worker count; `> 1` selects the concurrent
rendering (one aggregate spinner), `<= 1` the sequential one
(plain echoed lines).
:param spinner: a {class}`~click_extra.spinner_presets.SpinnerPreset`
from the {data}`~click_extra.spinner_presets.SPINNERS` catalog
(`spinner=SPINNERS["moon"]`) for the concurrent aggregate spinner.
Ignored by the sequential and progress-bar renderings, and mutually
exclusive with `progress_bar`.
:param progress_bar: render the aggregate indicator as a determinate
{func}`click.progressbar` instead of a spinner, for a sequential or
concurrent batch alike. Requires a positive `total` (a bar needs a
length) and is mutually exclusive with `spinner`.
:param timer: append each operation's and the batch's elapsed time to
the trail lines and the finisher. `None` (the default) follows the
CLI's `--time` / `--no-time` flag; `True` forces timing on with
{func}`~click_extra.humanize.format_duration`'s compact clock, a
callable `(seconds: float) -> str` forces it on with a custom
format, and `False` forces it off. Per-operation times come from a
`seconds` argument to {meth}`mark`, filled in automatically by an
{meth}`operation` handle.
:param clock: whether a running aggregate indicator shows *elapsed* time
(`"elapsed"`, the default: a stopwatch counting up, visible from the
start) or *remaining* time (`"eta"`: an estimate from the batch's
rate, appearing only once an outcome lets it be computed). Both the
progress bar and the concurrent spinner honor `"eta"` (the spinner
reuses Click's progress-bar estimate, since the trail knows its
`total`). Per-operation and finisher times are always elapsed.
:param visible: whether the trail shows anything. `False` silences the
lines, the finisher and the aggregate indicator, while
{attr}`ok_count` keeps counting.
:param live: where the aggregate indicator may draw: `"auto"` (the
default) on an interactive terminal only, unless `--no-progress`
turns it off, `"always"` on any stream, `"never"` nowhere. Where it
does not draw, each outcome line prints as it lands.
:param echo_sequential: whether the batch echoes its outcome lines and
finisher as plain lines at all: in a sequential batch, in a batch
whose aggregate indicator cannot draw on the stream, and in one that
finishes before its indicator first draws. Turn it off when the
batch has another output that is the real product (a result table)
and the trail would be noise. An indicator that did draw is
unaffected.
:param delay: seconds before the aggregate indicator first draws: a
fast batch then completes without ever flashing one, its lines
echoed plainly at {meth}`finish` instead (see `echo_sequential`).
:param stream: where to render; defaults to {data}`sys.stderr` so the
trail never mixes into `stdout` data.
:param enabled: deprecated, use `visible` and `live` instead: `False`
stands for `visible=False`, and `True` for `live="always"`.
:raises ValueError: if `progress_bar` is set without a positive
`total`, or together with `spinner`, if `clock` is neither
`"elapsed"` nor `"eta"`, or if `live` is not one of
{data}`~click_extra.spinner.LIVE_MODES`.
"""
if progress_bar and total <= 0:
raise ValueError("progress_bar=True requires a positive total.")
if progress_bar and spinner is not None:
raise ValueError("progress_bar= and spinner= are mutually exclusive.")
if clock not in ("elapsed", "eta"):
raise ValueError('clock must be "elapsed" or "eta".')
_check_live(live)
if enabled is not UNSET:
warn_deprecated_argument("OperationTrail", "enabled", "visible= and live=")
if enabled is False:
visible = False
elif enabled is True:
live = "always"
self.label = label
self.unit = unit
self.total = total
self.concurrent = jobs > 1
self.progress_bar = progress_bar
# None auto-detects the --time flag; a bool or callable forces it.
self.timer = _resolve_timer(timer)
self.clock = clock
self.spinner_preset = spinner
self.visible = visible
self.live = live
self.stream = stream
self._delay = delay
self._lock = threading.Lock()
self._done = 0
self._ok = 0
self._finished = False
self._start = time.monotonic()
self._indicator: _AggregateIndicator | None = None
self._buffer: list[str] = []
# Whether outcome lines may reach the stream as plain text: the trail is
# the batch's output and is visible. A plain line needs no cursor
# control, so this holds for a pipe or a file as for a terminal.
self._echo_plain = echo_sequential and visible
# The mode the aggregate indicator gets: a hidden trail draws nothing,
# whatever `live` says.
self._indicator_live: TLive = live if visible else "never"
# An aggregate indicator (a progress bar, or a spinner for a concurrent
# batch) owns the live line only on a stream it can draw on, under the
# same resolution the indicator applies. There, lines wait for its first
# frame, and finish() echoes them for a batch it never drew. Elsewhere
# no indicator ever draws, so lines echo as they are marked.
self._indicator_draws = (self.concurrent or progress_bar) and _can_draw(
self._indicator_live, _stream_or_stderr(stream)
)
self._echo = self._echo_plain and not self._indicator_draws
def __enter__(self) -> Self:
if self.progress_bar:
self._indicator = _BarIndicator(
label=self.label,
unit=self.unit,
total=self.total,
delay=self._delay,
live=self._indicator_live,
stream=self.stream,
timer=self.timer,
clock=self.clock,
)
elif self.concurrent:
self._indicator = _SpinnerIndicator(
label=self.label,
unit=self.unit,
total=self.total,
delay=self._delay,
live=self._indicator_live,
stream=self.stream,
spinner=self.spinner_preset,
timer=self.timer,
clock=self.clock,
)
if self._indicator is not None:
self._indicator.__enter__()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
if (
exc_type is not None
and issubclass(exc_type, KeyboardInterrupt)
and not self._finished
):
# An interrupted batch still closes with a finisher, so the record
# of what completed ends on a line saying it stopped short.
self.finish(False, self._interrupted_summary())
indicator = self._indicator
if indicator is None:
return
# Leaving without a finisher keeps every outcome line, as finish() does.
# Lines the indicator drew are already on screen, past any still buffered
# behind a first frame that fell due after the last mark. Lines of a batch
# it never drew echo plainly, on the same terms as in finish().
with self._lock:
if indicator.shown:
self._flush()
replay = self._buffer if self._echo_plain else []
self._buffer = []
indicator.__exit__(exc_type, exc_val, exc_tb)
self._indicator = None
for text in replay:
self._echo_line(text)
@property
def ok_count(self) -> int:
"""How many marked outcomes have succeeded so far."""
return self._ok
def _interrupted_summary(self) -> str:
"""The finisher of a batch a Ctrl+C stopped: how far it got."""
count = f"{self._done}/{self.total}" if self.total else str(self._done)
summary = _("Interrupted after {count}").format(count=count)
return f"{summary} {self.unit}" if self.unit else summary
def _render_line(self, ok: bool, message: str) -> str:
"""Format one `β`/`β` line for the trail's stream, plain when color is off.
The glyph is painted unconditionally by {func}`trail_line`, and a line
flushed above a drawn indicator reaches the stream through a raw write,
not through {func}`click.echo`'s strip. So the strip happens here, and
takes any escape embedded in `message` along with the glyph's.
"""
line = trail_line(ok, message)
stream = _stream_or_stderr(self.stream)
return line if _color_enabled(stream) else click.unstyle(line)
def _echo_line(self, message: str) -> None:
"""Print one rendered line to the trail's stream."""
if self.stream is not None:
click.echo(message, file=self.stream)
else:
click.echo(message, err=True)
[docs]
def mark(self, ok: bool, message: str, seconds: float | None = None) -> None:
"""Record one `β`/`β` outcome: tally it and render its trail line.
:param seconds: the operation's own elapsed time. When `timer` is on it
is formatted and appended to `message` as ` (2.3s)`. An
{meth}`operation` handle fills this in from when it was created;
pass it yourself when you already hold a duration.
"""
if self.timer and seconds is not None:
message = f"{message} ({_format_timer(self.timer, seconds)})"
with self._lock:
self._done += 1
if ok:
self._ok += 1
if self._finished:
# An outcome landing after the finisher, from a task the batch
# waited on after a Ctrl+C: print it plainly, below the finisher,
# so the record of what completed stays whole.
if self._echo_plain:
self._echo_line(self._render_line(ok, message))
elif self._echo:
self._echo_line(self._render_line(ok, message))
elif self._indicator is not None:
self._buffer.append(self._render_line(ok, message))
self._indicator.advance(self._done)
self._flush()
def _flush(self) -> None:
# Caller holds the lock. Drain buffered lines once the indicator is
# drawing; before that, writing would leak into a stream the delayed
# (or disabled) indicator may never touch.
if self._indicator is None or not self._indicator.shown:
return
for text in self._buffer:
self._indicator.echo(text)
self._buffer.clear()
[docs]
def finish(self, ok: bool, summary: str) -> None:
"""Render the persistent `β`/`β` ``{summary}`` finisher.
With an aggregate indicator that drew, it becomes the indicator's kept
line (a spinner's {meth}`Spinner.ok` / {meth}`Spinner.fail` line, or the
bar's replacement line); otherwise, a plain echoed line. The batch's
elapsed time since construction is appended when `timer` is on.
A batch finishing inside `delay` never draws its indicator, so none of
its buffered lines reached the stream. When the trail is the batch's
output (`echo_sequential` and `visible`), they are echoed plainly along
with the finisher, the way a sequential batch prints them: how fast a
batch ran must not decide whether its record exists.
"""
self._finished = True
indicator = self._indicator
if indicator is not None:
with self._lock:
drew = indicator.shown
if drew:
self._flush()
replay, self._buffer = self._buffer, []
if drew or not self._echo_plain:
indicator.finish(ok, summary)
return
# Tear the indicator down first, so a first frame falling due right
# now cannot draw over the replayed lines.
indicator.stop()
for text in replay:
self._echo_line(text)
elif not self._echo:
return
if self.timer:
elapsed = time.monotonic() - self._start
summary = f"{summary} ({_format_timer(self.timer, elapsed)})"
self._echo_line(self._render_line(ok, summary))
[docs]
def operation(self) -> _Operation:
"""Start a timed operation, returning a handle to record its outcome.
The handle captures the current time; call {meth}`_Operation.mark` when
the work finishes to record its `β`/`β` outcome with the elapsed time
appended (when `timer` is on). This is how a batch reports
per-operation timings under concurrency, where the trail itself never
sees when an operation began:
```{code-block} python
def fetch(feed):
op = trail.operation()
ok, message = pull(feed)
op.mark(ok, message)
```
"""
return _Operation(self)
class _Operation:
"""A single timed operation issued by {meth}`OperationTrail.operation`.
Captures its start time on creation; {meth}`mark` reports the outcome to the
parent trail with the elapsed time, so per-operation timings work under
concurrency where the trail cannot know when each operation began.
"""
def __init__(self, trail: OperationTrail) -> None:
self._trail = trail
self._start = time.monotonic()
def mark(self, ok: bool, message: str) -> None:
"""Record this operation's `β`/`β` outcome, timed from its start."""
self._trail.mark(ok, message, seconds=time.monotonic() - self._start)
[docs]
class ProgressOption(ExtraOption):
"""A pre-configured `--progress`/`--no-progress` flag gating spinner display.
Resolves to a single boolean published at
{data}`ctx.meta[click_extra.context.PROGRESS] <click_extra.context.PROGRESS>`,
which {class}`Spinner`, {class}`OperationTrail` and {func}`progressbar` read on
their own when left at their automatic default. The default is `True`;
`--accessible` lowers it to `False` (via `default_map`) so a screen reader is
never handed a spinning glyph.
```{note}
Spinner display is intentionally **decoupled from color**, even though both
emit ANSI. A spinner is an *interactivity* concern, not a color one: it is
built from cursor-control codes (hide-cursor, carriage return, clear-line),
which the [NO_COLOR standard](https://no-color.org) explicitly does not
govern -- it "only signals the user's intention regarding adding ANSI color
to text output". So `--no-color` / `NO_COLOR` strip the spinner's colors
but never hide it.
This matches how the wider ecosystem treats the two axes as orthogonal:
cargo, npm, pip, Rich, indicatif and ora all gate progress on the terminal
(and a dedicated `--progress`/`--quiet` knob), while `NO_COLOR` only
affects color. Rich uses `TERM=dumb` -- not `NO_COLOR` -- as the signal
to drop cursor-moving features like progress bars.
The spinner is therefore silenced by two things only, neither of them color:
- **non-interactive output** -- a pipe, file, CI log, or `TERM=dumb`
terminal that cannot move the cursor (see `Spinner._resolve_live`);
- **explicit intent** -- `--no-progress` or `--accessible`.
```
This option is eager. It no longer reads `ctx.color`, so its position relative
to {class}`~click_extra.color.ColorOption` is not load-bearing.
"""
[docs]
def set_progress(
self,
ctx: click.Context,
param: click.Parameter,
value: bool,
) -> None:
"""Publish whether progress spinners may be shown.
Stores the resolved `--progress` flag at
{data}`~click_extra.context.PROGRESS`. Deliberately independent of color:
see the {class}`ProgressOption` note for why a spinner is gated on
interactivity (TTY / `TERM=dumb`) and `--accessible`, never on
`--no-color` / `NO_COLOR`.
"""
context.set(ctx, context.PROGRESS, value)
def __init__(
self,
param_decls: Sequence[str] | None = None,
*,
is_flag: bool = True,
default: bool = True,
is_eager: bool = True,
expose_value: bool = False,
help: str = _(
"Show progress indicators during long operations. Disabled for "
"non-interactive output (pipes, dumb terminals, CI) and by --accessible."
),
**kwargs: Any,
) -> None:
if not param_decls:
param_decls = ("--progress/--no-progress",)
kwargs.setdefault("callback", self.set_progress)
super().__init__(
param_decls=param_decls,
is_flag=is_flag,
default=default,
is_eager=is_eager,
expose_value=expose_value,
help=help,
**kwargs,
)
V = TypeVar("V")
def _flush_final_position(bar: ProgressBar[Any]) -> None:
"""Make the bar render its true final position on finish.
Works around [pallets/click#3571](https://github.com/pallets/click/issues/3571):
Click's {meth}`~click._termui_impl.ProgressBar.update` only applies and
redraws accumulated steps once they reach `update_min_steps`, dropping the
trailing sub-threshold batch. A bar whose `length` is not a multiple of
`update_min_steps` (with `show_pos=True`) then freezes below completion:
`14/20`, not `20/20`, for `length=20` and `update_min_steps=7`. Wrapping
{meth}`render_finish` flushes those pending steps and redraws once more, so
the kept line shows the real final position. A no-op for the default
`update_min_steps=1`, where nothing is ever left pending.
"""
inner_render_finish = bar.render_finish
def render_finish() -> None:
pending = bar._completed_intervals
if pending:
bar.make_step(pending)
bar._completed_intervals = 0
bar.render_progress()
inner_render_finish()
bar.render_finish = render_finish # type: ignore[method-assign]
[docs]
def progressbar(
iterable: Iterable[V] | None = None,
length: int | None = None,
label: str | None = None,
hidden: bool | None = None,
show_eta: bool | None = None,
**kwargs: Any,
) -> ProgressBar[V]:
"""Drop-in for {func}`click.progressbar` honoring `--progress` and `--time`.
Click's own progress bar is *determinate*, the counterpart to the
indeterminate {class}`Spinner`. This thin wrapper gates its visibility on the
same {data}`~click_extra.context.PROGRESS` flag the spinner uses, so a single
`--no-progress` (or `--accessible`, which lowers the `progress` default)
silences both, and gates its estimated-time display on `--time`.
:param hidden: tri-state. Left at its default `None`, the bar follows the
resolved `--progress` flag: hidden when the user (or `--accessible`)
turned progress off, shown otherwise. An explicit `True` or `False`
forces the bar regardless, mirroring how an explicit `color=` argument
overrides `ctx.color` on {func}`click.echo`. With no active context (the
bar used outside a Click command) it defaults to shown.
:param show_eta: tri-state, like `hidden`. Left at its default `None`, the
estimated-time-remaining display follows the `--time` / `--no-time`
flag: shown under `--time`, hidden otherwise (its default, or outside a
command). An explicit `True` or `False` forces it, keeping a bare bar's
timing in step with an {class}`OperationTrail`'s `timer`. Click's own
default is `True`.
```{note}
The `--progress` flag gates visibility and `--time` the ETA. Color is
already handled upstream: Click renders the bar through {func}`click.echo`,
whose `color=None` resolves against `ctx.color`, so `--no-color` /
`NO_COLOR` strip the bar's ANSI without any work from this wrapper.
```
"""
if hidden is None:
hidden = not _progress_enabled()
if show_eta is None:
# The ETA follows --time / --no-time, like a trail's timer, so a bare
# bar and a trail agree on when to show timing.
show_eta = _time_flag_active()
bar = click.progressbar(
iterable,
length=length,
label=label,
hidden=hidden,
show_eta=show_eta,
**kwargs,
)
# Repair the final-position freeze of pallets/click#3571 (harmless otherwise).
_flush_final_position(bar)
return bar
# Max display width (terminal cells) of the frame preview column.
_SPINNER_PREVIEW_WIDTH = 56
def _spinner_preview(preset: SpinnerPreset) -> str:
"""Join leading frames into a preview within the display-width budget.
Frames are measured by terminal cell width
({func}`~click_extra.layout.cell_width`), not by code points, so 1-cell
glyphs and 2-cell emoji fill the column consistently rather than letting an
emoji-heavy preview balloon it. Emoji variation selectors (`U+FE0F`) are
dropped: `wcwidth` sizes the promoted emoji at two cells while many
terminals render the bare symbol in one, and that disagreement misaligns
the table. Wide animations (`shark`, `pong`, `dots-8bit`, β¦) stop at the
budget with a `β¦ (+N)` tail.
"""
shown: list[str] = []
width = 0
for frame in preset.frames:
glyph = frame.replace("\ufe0f", "") # Drop emoji variation selectors.
cost = cell_width(glyph) + (1 if shown else 0) # +1 joining space.
if width + cost > _SPINNER_PREVIEW_WIDTH:
break
shown.append(glyph)
width += cost
preview = " ".join(shown)
remaining = len(preset.frames) - len(shown)
if remaining:
preview += f" β¦ (+{remaining})"
return preview
# A curated, visually-distinct default selection for the live tour.
_DEFAULT_SHOWCASE = (
"dots",
"line",
"moon",
"clock",
"earth",
"bouncing-bar",
"arc",
"pong",
"shark",
"mindblown",
)
# The live tour aims for _TOUR_CYCLES full cycles per spinner, then bounds the
# dwell to at least _TOUR_MIN seconds (so a snappy spinner stays watchable) and
# at most _TOUR_CAP seconds (so a long or slow one does not monopolize the tour).
_TOUR_CYCLES = 3
_TOUR_MIN = 2.0
_TOUR_CAP = 3.0
def _tour_duration(preset: SpinnerPreset) -> float:
"""Seconds the live tour dwells on a spinner.
Aims for {data}`_TOUR_CYCLES` full cycles (one cycle is a pass through every
frame), then clamps to `[_TOUR_MIN, _TOUR_CAP]` seconds: a snappy spinner is
held at least {data}`_TOUR_MIN` seconds so it is watchable, while a long or
slow one is capped at {data}`_TOUR_CAP`. The cap never trims below a single
full cycle, so even a 256-frame spinner completes one loop.
"""
one_cycle = len(preset.frames) * preset.interval
capped = min(_TOUR_CYCLES * one_cycle, max(_TOUR_CAP, one_cycle))
return max(_TOUR_MIN, capped)
def _animate_spinners(names: list[str]) -> None:
"""Spin each named catalog animation live, with its label and elapsed timer.
Each spinner runs for its {func}`_tour_duration` (up to {data}`_TOUR_CYCLES`
cycles, capped at {data}`_TOUR_CAP` seconds) before moving on, then leaves a
`β` success line behind. Interactive terminals only.
"""
for name in names:
preset = SPINNERS[name]
with Spinner(name, spinner=preset, timer=True) as spinner:
time.sleep(_tour_duration(preset))
spinner.ok()