Coverage for typer / _click / _termui_impl.py: 100%
289 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-11-09 00:15 +0000
1"""
2To keep the import times down, some infrequently used termui functionality
3is placed here and only imported as needed.
4"""
6import contextlib 1abfgcde
7import math 1abfgcde
8import os 1abfgcde
9import sys 1abfgcde
10import time 1abfgcde
11from collections.abc import Callable, Generator, Iterable, Iterator 1abfgcde
12from io import StringIO 1abfgcde
13from types import TracebackType 1abfgcde
14from typing import Generic, TextIO, TypeVar, cast 1abfgcde
16from ._compat import ( 1abfgcde
17 CYGWIN,
18 WIN,
19 _default_text_stdout,
20 get_best_encoding,
21 isatty,
22 term_len,
23)
24from .utils import echo 1abfgcde
26V = TypeVar("V") 1abfgcde
28if os.name == "nt": 1abfgcde
29 BEFORE_BAR = "\r" 1fg
30 AFTER_BAR = "\n" 1fg
31else:
32 BEFORE_BAR = "\r\033[?25l" 1abcde
33 AFTER_BAR = "\033[?25h\n" 1abcde
36class ProgressBar(Generic[V]): 1abfgcde
37 def __init__( 1abfgcde
38 self,
39 iterable: Iterable[V] | None,
40 length: int | None = None,
41 fill_char: str = "#",
42 empty_char: str = " ",
43 bar_template: str = "%(bar)s",
44 info_sep: str = " ",
45 hidden: bool = False,
46 show_eta: bool = True,
47 show_percent: bool | None = None,
48 show_pos: bool = False,
49 item_show_func: Callable[[V | None], str | None] | None = None,
50 label: str | None = None,
51 file: TextIO | None = None,
52 color: bool | None = None,
53 update_min_steps: int = 1,
54 width: int = 30,
55 ) -> None:
56 self.fill_char = fill_char 1abfgcde
57 self.empty_char = empty_char 1abfgcde
58 self.bar_template = bar_template 1abfgcde
59 self.info_sep = info_sep 1abfgcde
60 self.hidden = hidden 1abfgcde
61 self.show_eta = show_eta 1abfgcde
62 self.show_percent = show_percent 1abfgcde
63 self.show_pos = show_pos 1abfgcde
64 self.item_show_func = item_show_func 1abfgcde
65 self.label: str = label or "" 1abfgcde
67 if file is None: 1abfgcde
68 file = _default_text_stdout() 1abfgcde
70 # There are no standard streams attached to write to. For example,
71 # pythonw on Windows.
72 if file is None: # pragma: no cover 1abfgcde
73 file = StringIO()
75 self.file = file 1abfgcde
76 self.color = color 1abfgcde
77 self.update_min_steps = update_min_steps 1abfgcde
78 self._completed_intervals = 0 1abfgcde
79 self.width: int = width 1abfgcde
80 self.autowidth: bool = width == 0 1abfgcde
82 if length is None: 1abfgcde
83 from operator import length_hint 1abfgcde
85 length = length_hint(iterable, -1) 1abfgcde
87 if length == -1: # pragma: no cover 1abfgcde
88 length = None 1abfgcde
89 if iterable is None: 1abfgcde
90 if length is None: # pragma: no cover 1abfgcde
91 raise TypeError("iterable or length is required")
92 iterable = cast("Iterable[V]", range(length)) 1abfgcde
93 self.iter: Iterable[V] = iter(iterable) 1abfgcde
94 self.length = length 1abfgcde
95 self.pos: int = 0 1abfgcde
96 self.avg: list[float] = [] 1abfgcde
97 self.last_eta: float 1abfgcde
98 self.start: float 1abfgcde
99 self.start = self.last_eta = time.time() 1abfgcde
100 self.eta_known: bool = False 1abfgcde
101 self.finished: bool = False 1abfgcde
102 self.max_width: int | None = None 1abfgcde
103 self.entered: bool = False 1abfgcde
104 self.current_item: V | None = None 1abfgcde
105 self._is_atty = isatty(self.file) 1abfgcde
106 self._last_line: str | None = None 1abfgcde
108 def __enter__(self) -> "ProgressBar[V]": 1abfgcde
109 self.entered = True 1abfgcde
110 self.render_progress() 1abfgcde
111 return self 1abfgcde
113 def __exit__( 1abfgcde
114 self,
115 exc_type: type[BaseException] | None,
116 exc_value: BaseException | None,
117 tb: TracebackType | None,
118 ) -> None:
119 self.render_finish() 1abfgcde
121 def __iter__(self) -> Iterator[V]: 1abfgcde
122 if not self.entered: 1abfgcde
123 raise RuntimeError("You need to use progress bars in a with block.") 1abfgcde
124 self.render_progress() 1abfgcde
125 return self.generator() 1abfgcde
127 def __next__(self) -> V: 1abfgcde
128 # Iteration is defined in terms of a generator function,
129 # returned by iter(self); use that to define next(). This works
130 # because `self.iter` is an iterable consumed by that generator,
131 # so it is re-entry safe. Calling `next(self.generator())`
132 # twice works and does "what you want".
133 return next(iter(self)) 1abfgcde
135 def render_finish(self) -> None: 1abfgcde
136 if self.hidden or not self._is_atty: 1abfgcde
137 return 1abfgcde
138 self.file.write(AFTER_BAR) 1abfgcde
139 self.file.flush() 1abfgcde
141 @property 1abfgcde
142 def pct(self) -> float: 1abfgcde
143 if self.finished: 1abfgcde
144 return 1.0 1abfgcde
145 return min(self.pos / (float(self.length or 1) or 1), 1.0) 1abfgcde
147 @property 1abfgcde
148 def time_per_iteration(self) -> float: 1abfgcde
149 if not self.avg: 1abfgcde
150 return 0.0 1abfgcde
151 return sum(self.avg) / float(len(self.avg)) 1abfgcde
153 @property 1abfgcde
154 def eta(self) -> float: 1abfgcde
155 if self.length is not None and not self.finished: 1abfgcde
156 return self.time_per_iteration * (self.length - self.pos) 1abfgcde
157 return 0.0 1abfgcde
159 def format_eta(self) -> str: 1abfgcde
160 if self.eta_known: 1abfgcde
161 t = int(self.eta) 1abfgcde
162 seconds = t % 60 1abfgcde
163 t //= 60 1abfgcde
164 minutes = t % 60 1abfgcde
165 t //= 60 1abfgcde
166 hours = t % 24 1abfgcde
167 t //= 24 1abfgcde
168 if t > 0: 1abfgcde
169 return f"{t}d {hours:02}:{minutes:02}:{seconds:02}" 1abfgcde
170 else:
171 return f"{hours:02}:{minutes:02}:{seconds:02}" 1abfgcde
172 return "" 1abfgcde
174 def format_pos(self) -> str: 1abfgcde
175 pos = str(self.pos) 1abfgcde
176 if self.length is not None: 1abfgcde
177 pos += f"/{self.length}" 1abfgcde
178 return pos 1abfgcde
180 def format_pct(self) -> str: 1abfgcde
181 return f"{int(self.pct * 100): 4}%"[1:] 1abfgcde
183 def format_bar(self) -> str: 1abfgcde
184 if self.length is not None: 1abfgcde
185 bar_length = int(self.pct * self.width) 1abfgcde
186 bar = self.fill_char * bar_length 1abfgcde
187 bar += self.empty_char * (self.width - bar_length) 1abfgcde
188 elif self.finished: 1abfgcde
189 bar = self.fill_char * self.width 1abfgcde
190 else:
191 chars = list(self.empty_char * (self.width or 1)) 1abfgcde
192 if self.time_per_iteration != 0: 1abfgcde
193 chars[ 1abfgcde
194 int(
195 (math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
196 * self.width
197 )
198 ] = self.fill_char
199 bar = "".join(chars) 1abfgcde
200 return bar 1abfgcde
202 def format_progress_line(self) -> str: 1abfgcde
203 show_percent = self.show_percent 1abfgcde
205 info_bits = [] 1abfgcde
206 if self.length is not None and show_percent is None: 1abfgcde
207 show_percent = not self.show_pos 1abfgcde
209 if self.show_pos: 1abfgcde
210 info_bits.append(self.format_pos()) 1abfgcde
211 if show_percent: 1abfgcde
212 info_bits.append(self.format_pct()) 1abfgcde
213 if self.show_eta and self.eta_known and not self.finished: 1abfgcde
214 info_bits.append(self.format_eta()) 1abfgcde
215 if self.item_show_func is not None: 1abfgcde
216 item_info = self.item_show_func(self.current_item) 1abfgcde
217 if item_info is not None: 1abfgcde
218 info_bits.append(item_info) 1abfgcde
220 return ( 1abfgcde
221 self.bar_template
222 % {
223 "label": self.label,
224 "bar": self.format_bar(),
225 "info": self.info_sep.join(info_bits),
226 }
227 ).rstrip()
229 def render_progress(self) -> None: 1abfgcde
230 if self.hidden: 1abfgcde
231 return 1abfgcde
233 if not self._is_atty: 1abfgcde
234 # Only output the label once if the output is not a TTY.
235 if self._last_line != self.label: 1abfgcde
236 self._last_line = self.label 1abfgcde
237 echo(self.label, file=self.file, color=self.color) 1abfgcde
238 return 1abfgcde
240 buf = [] 1abfgcde
241 # Update width in case the terminal has been resized
242 if self.autowidth: 1abfgcde
243 import shutil 1abfgcde
245 old_width = self.width 1abfgcde
246 self.width = 0 1abfgcde
247 clutter_length = term_len(self.format_progress_line()) 1abfgcde
248 new_width = max(0, shutil.get_terminal_size().columns - clutter_length) 1abfgcde
249 if new_width < old_width and self.max_width is not None: 1abfgcde
250 buf.append(BEFORE_BAR) 1abfgcde
251 buf.append(" " * self.max_width) 1abfgcde
252 self.max_width = new_width 1abfgcde
253 self.width = new_width 1abfgcde
255 clear_width = self.width 1abfgcde
256 if self.max_width is not None: 1abfgcde
257 clear_width = self.max_width 1abfgcde
259 buf.append(BEFORE_BAR) 1abfgcde
260 line = self.format_progress_line() 1abfgcde
261 line_len = term_len(line) 1abfgcde
262 if self.max_width is None or self.max_width < line_len: 1abfgcde
263 self.max_width = line_len 1abfgcde
265 buf.append(line) 1abfgcde
266 buf.append(" " * (clear_width - line_len)) 1abfgcde
267 line = "".join(buf) 1abfgcde
268 # Render the line only if it changed.
270 if line != self._last_line: 1abfgcde
271 self._last_line = line 1abfgcde
272 echo(line, file=self.file, color=self.color, nl=False) 1abfgcde
273 self.file.flush() 1abfgcde
275 def make_step(self, n_steps: int) -> None: 1abfgcde
276 self.pos += n_steps 1abfgcde
277 if self.length is not None and self.pos >= self.length: 1abfgcde
278 self.finished = True 1abfgcde
280 if (time.time() - self.last_eta) < 1.0: 1abfgcde
281 return 1abfgcde
283 self.last_eta = time.time() 1abfgcde
285 # self.avg is a rolling list of length <= 7 of steps where steps are
286 # defined as time elapsed divided by the total progress through
287 # self.length.
288 if self.pos: 1abfgcde
289 step = (time.time() - self.start) / self.pos 1abfgcde
290 else:
291 step = time.time() - self.start 1abfgcde
293 self.avg = self.avg[-6:] + [step] 1abfgcde
295 self.eta_known = self.length is not None 1abfgcde
297 def update(self, n_steps: int) -> None: 1abfgcde
298 """Update the progress bar by advancing a specified number of steps."""
299 self._completed_intervals += n_steps 1abfgcde
301 if self._completed_intervals >= self.update_min_steps: 1abfgcde
302 self.make_step(self._completed_intervals) 1abfgcde
303 self.render_progress() 1abfgcde
304 self._completed_intervals = 0 1abfgcde
306 def finish(self) -> None: 1abfgcde
307 self.eta_known = False 1abfgcde
308 self.current_item = None 1abfgcde
309 self.finished = True 1abfgcde
311 def generator(self) -> Iterator[V]: 1abfgcde
312 """Return a generator which yields the items added to the bar
313 during construction, and updates the progress bar *after* the
314 yielded block returns.
315 """
316 # WARNING: the iterator interface for `ProgressBar` relies on
317 # this and only works because this is a simple generator which
318 # doesn't create or manage additional state. If this function
319 # changes, the impact should be evaluated both against
320 # `iter(bar)` and `next(bar)`. `next()` in particular may call
321 # `self.generator()` repeatedly, and this must remain safe in
322 # order for that interface to work.
323 if not self.entered: # pragma: no cover 1abfgcde
324 raise RuntimeError("You need to use progress bars in a with block.")
326 if not self._is_atty: 1abfgcde
327 yield from self.iter 1abfgcde
328 else:
329 for rv in self.iter: 1abfgcde
330 self.current_item = rv 1abfgcde
332 # This allows show_item_func to be updated before the
333 # item is processed. Only trigger at the beginning of
334 # the update interval.
335 if self._completed_intervals == 0: 1abfgcde
336 self.render_progress() 1abfgcde
338 yield rv 1abfgcde
339 self.update(1) 1abfgcde
341 self.finish() 1abfgcde
342 self.render_progress() 1abfgcde
345def open_url(url: str, wait: bool = False, locate: bool = False) -> int: 1abfgcde
346 import subprocess 1abfgcde
348 def _unquote_file(url: str) -> str: 1abfgcde
349 from urllib.parse import unquote 1abfgcde
351 if url.startswith("file://"): 1abfgcde
352 url = unquote(url[7:]) 1fg
354 return url 1abfgcde
356 if sys.platform == "darwin": 1abfgcde
357 args = ["open"] 1cde
358 if wait: 1cde
359 args.append("-W") 1cde
360 if locate: 1cde
361 args.append("-R") 1cde
362 args.append(_unquote_file(url)) 1cde
363 null = open("/dev/null", "w") 1cde
364 try: 1cde
365 return subprocess.Popen(args, stderr=null).wait() 1cde
366 finally:
367 null.close() 1cde
368 elif WIN: 1abfg
369 if locate: 1fg
370 url = _unquote_file(url) 1fg
371 args = ["explorer", f"/select,{url}"] 1fg
372 else:
373 args = ["start"] 1fg
374 if wait: 1fg
375 args.append("/WAIT") 1fg
376 args.append("") 1fg
377 args.append(url) 1fg
378 try: 1fg
379 return subprocess.call(args) 1fg
380 except OSError: 1fg
381 # Command not found
382 return 127 1fg
383 elif CYGWIN: # pragma: no cover 1ab
384 if locate:
385 url = _unquote_file(url)
386 args = ["cygstart", os.path.dirname(url)]
387 else:
388 args = ["cygstart"]
389 if wait:
390 args.append("-w")
391 args.append(url)
392 try:
393 return subprocess.call(args)
394 except OSError:
395 # Command not found
396 return 127
398 try: 1ab
399 if locate: 1ab
400 url = os.path.dirname(_unquote_file(url)) or "." 1ab
401 else:
402 url = _unquote_file(url) 1ab
403 c = subprocess.Popen(["xdg-open", url]) 1ab
404 if wait: 1ab
405 return c.wait() 1ab
406 return 0 1ab
407 except OSError: # pragma: no cover
408 # TODO: remove this part, doesn't get hit by Typer code paths?
409 if url.startswith(("http://", "https://")) and not locate and not wait:
410 import webbrowser
412 webbrowser.open(url)
413 return 0
414 return 1
417def _translate_ch_to_exc(ch: str) -> None: 1abfgcde
418 if ch == "\x03": 1abfgcde
419 raise KeyboardInterrupt() 1fg
421 if ch == "\x04" and not WIN: # Unix-like, Ctrl+D 1abfgcde
422 raise EOFError() 1abcde
424 if ch == "\x1a" and WIN: # Windows, Ctrl+Z 1abfgcde
425 raise EOFError() 1fg
427 return None 1abfgcde
430if sys.platform == "win32": 1abfgcde
431 import msvcrt 1fg
433 @contextlib.contextmanager 1fg
434 def raw_terminal() -> Iterator[int]: 1fg
435 yield -1 1fg
437 def getchar(echo: bool) -> str: 1fg
438 # The function `getch` will return a bytes object corresponding to
439 # the pressed character. Since Windows 10 build 1803, it will also
440 # return \x00 when called a second time after pressing a regular key.
441 #
442 # `getwch` does not share this probably-bugged behavior. Moreover, it
443 # returns a Unicode object by default, which is what we want.
444 #
445 # Either of these functions will return \x00 or \xe0 to indicate
446 # a special key, and you need to call the same function again to get
447 # the "rest" of the code. The fun part is that \u00e0 is
448 # "latin small letter a with grave", so if you type that on a French
449 # keyboard, you _also_ get a \xe0.
450 # E.g., consider the Up arrow. This returns \xe0 and then \x48. The
451 # resulting Unicode string reads as "a with grave" + "capital H".
452 # This is indistinguishable from when the user actually types
453 # "a with grave" and then "capital H".
454 #
455 # When \xe0 is returned, we assume it's part of a special-key sequence
456 # and call `getwch` again, but that means that when the user types
457 # the \u00e0 character, `getchar` doesn't return until a second
458 # character is typed.
459 # The alternative is returning immediately, but that would mess up
460 # cross-platform handling of arrow keys and others that start with
461 # \xe0. Another option is using `getch`, but then we can't reliably
462 # read non-ASCII characters, because return values of `getch` are
463 # limited to the current 8-bit codepage.
464 #
465 # Anyway, Click doesn't claim to do this Right(tm), and using `getwch`
466 # is doing the right thing in more situations than with `getch`.
468 if echo: 1fg
469 func = cast(Callable[[], str], msvcrt.getwche) 1fg
470 else:
471 func = cast(Callable[[], str], msvcrt.getwch) 1fg
473 rv = func() 1fg
475 if rv in ("\x00", "\xe0"): 1fg
476 # \x00 and \xe0 are control characters that indicate special key,
477 # see above.
478 rv += func() 1fg
480 _translate_ch_to_exc(rv) 1fg
481 return rv 1fg
483else:
484 import termios 1abcde
485 import tty 1abcde
487 @contextlib.contextmanager 1abcde
488 def raw_terminal() -> Generator[int]: 1abcde
489 f: TextIO | None
490 fd: int
492 if not isatty(sys.stdin): 1abcde
493 f = open("/dev/tty") 1abcde
494 fd = f.fileno() 1abcde
495 else:
496 fd = sys.stdin.fileno() 1abcde
497 f = None 1abcde
499 try: 1abcde
500 old_settings = termios.tcgetattr(fd) 1abcde
502 try: 1abcde
503 tty.setraw(fd) 1abcde
504 yield fd 1abcde
505 finally:
506 termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) 1abcde
507 sys.stdout.flush() 1abcde
509 if f is not None: 1abcde
510 f.close() 1abcde
511 except termios.error: # pragma: no cover
512 pass
514 def getchar(echo: bool) -> str: 1abcde
515 with raw_terminal() as fd: 1abcde
516 ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace") 1abcde
518 if echo and isatty(sys.stdout): # pragma: no cover 1abcde
519 sys.stdout.write(ch) 1abcde
521 _translate_ch_to_exc(ch) 1abcde
522 return ch 1abcde