# 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.
```
This module also keeps the registry of the live lines drawing on the terminal,
which {class}`~click_extra.trail.OperationTrail` and
{func}`~click_extra.progress.progressbar` join while they draw.
```{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 itertools
import os
import signal
import sys
import threading
import time
from gettext import gettext as _
import click
from . import context
from ._utils import AUTO
from .color import (
WHEN,
invocation_color,
is_a_tty,
is_dumb_terminal,
resolve_color_env,
)
from .humanize import format_duration
from .parameters import ExtraOption
from .spinner_presets import (
SPINNER_FRAMES,
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, Sequence
from types import FrameType, TracebackType
from typing import IO, Any, Final, Protocol, Self
from .color import TWhen
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}`~click_extra.trail.OperationTrail`'s progress-bar indicator.
"""
def _resolve_stream(self) -> IO[str]: ...
def echo(self, message: str) -> 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."""
CLEAR_LINE: Final = "\x1b[K"
"""Erase the line from the cursor to its end.
Written after a carriage return, it clears the row a live line is about to
redraw, so a shrinking label leaves no stale characters behind.
"""
HIDE_CURSOR: Final = "\x1b[?25l"
"""Hide the terminal's cursor, which would otherwise flicker beside a live line."""
SHOW_CURSOR: Final = "\x1b[?25h"
"""Show the cursor again, once a live line stops drawing."""
DEFAULT_INTERVAL: Final = 0.1
"""Seconds between two frames of a {class}`Spinner` given no other cadence.
Also how often a progress bar carrying a running clock redraws, so the two live
lines tick at the same rate.
"""
_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.
"""
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(f"\r{CLEAR_LINE}{SHOW_CURSOR}")
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.
"""
# Deferred import: keeps extra_platforms off the startup path of every CLI.
from extra_platforms import is_windows
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 _innermost_line(
stream: IO[str] | None,
accept: Callable[[_LiveLine], bool],
) -> _LiveLine | None:
"""Scan the live lines, innermost first, for one `accept` takes.
With `stream` given, only a line drawing on that very stream matches. The
one locked walk of {data}`_ACTIVE_LINES` both {func}`_active_line` and
{func}`active_spinner` perform.
"""
with _ACTIVE_LINES_LOCK:
for line in reversed(_ACTIVE_LINES):
if accept(line) and (stream is None or line._resolve_stream() is stream):
return line
return None
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.
"""
return _innermost_line(stream, lambda line: True)
[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.
"""
line = _innermost_line(stream, lambda line: isinstance(line, Spinner))
assert line is None or isinstance(line, Spinner)
return line
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)
def _plain_unless(line: str, color: bool) -> str:
"""Return `line` as it is when `color` is on, stripped of its escapes otherwise.
A line reaching the stream through a raw write, rather than through
{func}`click.echo`, has to drop its own escapes where color is off. The
strip takes any escape embedded in a label along with the ones the line
added itself.
"""
return line if color else click.unstyle(line)
def _paint(text: str, style: Style | None, color: bool) -> str:
"""Apply `style` to `text` when `color` is on and there is text to paint.
:param text: what to paint.
:param style: the style to apply, or `None` to leave `text` bare.
:param color: whether any color reaches the output at all.
:return: the text, styled or bare.
"""
if color and style is not None and text:
return style(text)
return text
def _check_live(live: str) -> None:
"""Reject a `live` value outside {data}`~click_extra.color.WHEN`.
:raises ValueError: if `live` is not one of {data}`~click_extra.color.WHEN`.
"""
if live not in WHEN:
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 is_dumb_terminal():
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
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 ANSI control codes of a live
line literally (`⠋␛[0m … ␛[K`) instead of animating in place. Modern
terminals (Windows Terminal, recent conhost) already enable it. 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 or 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
def _halt(thread: threading.Thread, event: threading.Event) -> bool:
"""Ask a drawing thread to stop, and wait {data}`_STOP_GRACE` for it.
:param thread: the daemon thread drawing the live line.
:param event: the stop event its loop waits on.
:return: whether the thread stopped. `False` means it is still blocked in a
write, so the caller skips its own clean-up writes, see
{data}`_STOP_GRACE`.
"""
event.set()
thread.join(timeout=_STOP_GRACE)
return not thread.is_alive()
[docs]
def trail_glyph(
ok: bool,
symbol: str | None = None,
style: Style | None = None,
*,
color: bool = True,
) -> str:
"""Return the `✓` or `✘` glyph marking an outcome, painted for it.
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.
:param ok: whether the outcome is a success.
:param symbol: a glyph to draw in place of the themed one.
:param style: a style to paint the glyph with, in place of the theme's slot.
:param color: whether to paint the glyph at all. With color off, the glyph
goes bare and the theme is never consulted.
:return: the glyph, painted or bare.
"""
glyph = symbol if symbol is not None else (OK_GLYPH if ok else KO_GLYPH)
if not color:
return glyph
paint: Callable[[str], str]
if style is None:
theme = get_current_theme()
paint = theme.success if ok else theme.error
else:
paint = style
return paint(glyph)
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 _timer_suffix(timer: bool | Callable[[float], str], seconds: float) -> str:
"""The ` (elapsed)` suffix a timed line ends on, or empty when timing is off.
The one spelling of the clock every live line and finisher appends: a
space, then the duration {func}`_format_timer` writes, in parentheses.
"""
if not timer:
return ""
return f" ({_format_timer(timer, 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
[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. A tool
running its steps one after another can share one instance and reassign its
{attr}`label` between steps. Concurrent work takes 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: TWhen = AUTO,
hide_cursor: bool = True,
beep: bool = False,
) -> 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
{data}`~click_extra.spinner.DEFAULT_INTERVAL`, 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}`~click_extra.trail.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.
: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.color.WHEN`.
"""
# A bare `@Spinner` decorator (no parentheses) passes the wrapped function
# as the first positional, where `@Spinner(…)` and `with Spinner(…)` pass
# a label. A string is never callable, so this never misfires on a label.
#
# `click_extra.decorators.allow_missing_parenthesis` runs the same test,
# but it wraps a decorator factory and returns a function, which would
# replace this class. `Spinner` stays a class to double as a context
# manager and to support `isinstance()`, so the parenthesis-less form is
# caught here, the one place it 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 its 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 None:
frames = SPINNER_FRAMES if spinner is None else spinner.frames
if interval is None:
interval = DEFAULT_INTERVAL if spinner is None else spinner.interval
self.frames = frames
self.interval = interval
_check_live(live)
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
# Whether the animation paints its colors, resolved by start(). It does
# not decide whether the spinner animates: a spinner spins in plain text
# on a terminal under NO_COLOR, as the ProgressOption note explains.
self._color = 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)
@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:
```{code-block} python
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, see `_timer_suffix`.
"""
return _timer_suffix(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: whether to paint the line. `None` follows what {meth}`start`
resolved for the spinner's own stream, which is what the animation
wants. A picture of the spinner states its own answer.
:return: the line, ANSI escape sequences included when color is on.
"""
if color is None:
color = self._color
label = f" {self.label}" if self.label else ""
clock = self._clock()
if self.label_style is None and self.timer_style is None:
line = _paint(f"{frame}{label}{clock}", self.style, color)
else:
label_style = self.style if self.label_style is None else self.label_style
timer_style = self.style if self.timer_style is None else self.timer_style
line = (
_paint(frame, self.style, color)
+ _paint(label, label_style, color)
+ _paint(clock, timer_style, color)
)
return _plain_unless(line, color)
[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
)
[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 `_color_enabled` reads.
self._start_time = time.monotonic()
self._stop_time = None
stream = self._resolve_stream()
self._color = _color_enabled(stream)
if not _can_draw(self.live, stream):
return
# The spinner is about to emit ANSI control codes: make sure a Windows
# console will interpret rather than echo them.
_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
halted = _halt(self._thread, self._stop)
self._thread = None
if not halted:
# 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 += f"\r{CLEAR_LINE}"
if self._cursor_hidden:
cleanup += SHOW_CURSOR
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(f"\r{CLEAR_LINE}")
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.
"""
stream = self._resolve_stream()
color = _color_enabled(stream)
self.stop()
label = _paint(f" {self.label}" if self.label else "", self.label_style, color)
clock = _paint(self._clock(), self.timer_style, color)
marker = trail_glyph(success, symbol, style, color=color)
line = _plain_unless(f"{marker}{label}{clock}", color)
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
try:
if self.hide_cursor:
stream.write(HIDE_CURSOR)
self._cursor_hidden = True
stream.flush()
for frame in itertools.cycle(self._ordered_frames):
if self._stop.is_set():
break
content = self._compose_frame(frame)
# 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}{CLEAR_LINE}")
stream.flush()
self._drawn = True
self._stop.wait(self.interval)
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]
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}`~click_extra.trail.OperationTrail` and
{func}`~click_extra.progress.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 **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) 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.
The wider ecosystem treats the two axes the same way: 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
takes `TERM=dumb`, not `NO_COLOR`, as the signal to drop cursor-moving
features like progress bars.
Two things silence the spinner, neither of them color:
- **non-interactive output**: a pipe, file, CI log, or `TERM=dumb` terminal
that cannot move the cursor (see {func}`_can_draw`);
- **explicit intent**: `--no-progress` or `--accessible`.
```
This option is eager. It never reads `ctx.color`, so its position relative to
{class}`~click_extra.color.ColorOption` does not matter.
"""
[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,
)