Загрузить файлы в «venv/Lib/site-packages/click»
This commit is contained in:
126
venv/Lib/site-packages/click/__init__.py
Normal file
126
venv/Lib/site-packages/click/__init__.py
Normal file
@@ -0,0 +1,126 @@
|
||||
"""
|
||||
Click is a simple Python module inspired by the stdlib optparse to make
|
||||
writing command line scripts fun. Unlike other modules, it's based
|
||||
around a simple API that does not come with too much magic and is
|
||||
composable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .core import Argument as Argument
|
||||
from .core import Command as Command
|
||||
from .core import CommandCollection as CommandCollection
|
||||
from .core import Context as Context
|
||||
from .core import Group as Group
|
||||
from .core import Option as Option
|
||||
from .core import Parameter as Parameter
|
||||
from .core import ParameterSource as ParameterSource
|
||||
from .decorators import argument as argument
|
||||
from .decorators import command as command
|
||||
from .decorators import confirmation_option as confirmation_option
|
||||
from .decorators import group as group
|
||||
from .decorators import help_option as help_option
|
||||
from .decorators import make_pass_decorator as make_pass_decorator
|
||||
from .decorators import option as option
|
||||
from .decorators import pass_context as pass_context
|
||||
from .decorators import pass_obj as pass_obj
|
||||
from .decorators import password_option as password_option
|
||||
from .decorators import version_option as version_option
|
||||
from .exceptions import Abort as Abort
|
||||
from .exceptions import BadArgumentUsage as BadArgumentUsage
|
||||
from .exceptions import BadOptionUsage as BadOptionUsage
|
||||
from .exceptions import BadParameter as BadParameter
|
||||
from .exceptions import ClickException as ClickException
|
||||
from .exceptions import FileError as FileError
|
||||
from .exceptions import MissingParameter as MissingParameter
|
||||
from .exceptions import NoSuchCommand as NoSuchCommand
|
||||
from .exceptions import NoSuchOption as NoSuchOption
|
||||
from .exceptions import UsageError as UsageError
|
||||
from .formatting import HelpFormatter as HelpFormatter
|
||||
from .formatting import wrap_text as wrap_text
|
||||
from .globals import get_current_context as get_current_context
|
||||
from .termui import clear as clear
|
||||
from .termui import confirm as confirm
|
||||
from .termui import echo_via_pager as echo_via_pager
|
||||
from .termui import edit as edit
|
||||
from .termui import get_pager_file as get_pager_file
|
||||
from .termui import getchar as getchar
|
||||
from .termui import launch as launch
|
||||
from .termui import pause as pause
|
||||
from .termui import progressbar as progressbar
|
||||
from .termui import prompt as prompt
|
||||
from .termui import secho as secho
|
||||
from .termui import style as style
|
||||
from .termui import unstyle as unstyle
|
||||
from .types import BOOL as BOOL
|
||||
from .types import Choice as Choice
|
||||
from .types import DateTime as DateTime
|
||||
from .types import File as File
|
||||
from .types import FLOAT as FLOAT
|
||||
from .types import FloatRange as FloatRange
|
||||
from .types import INT as INT
|
||||
from .types import IntRange as IntRange
|
||||
from .types import ParamType as ParamType
|
||||
from .types import Path as Path
|
||||
from .types import STRING as STRING
|
||||
from .types import Tuple as Tuple
|
||||
from .types import UNPROCESSED as UNPROCESSED
|
||||
from .types import UUID as UUID
|
||||
from .utils import echo as echo
|
||||
from .utils import format_filename as format_filename
|
||||
from .utils import get_app_dir as get_app_dir
|
||||
from .utils import get_binary_stream as get_binary_stream
|
||||
from .utils import get_text_stream as get_text_stream
|
||||
from .utils import open_file as open_file
|
||||
|
||||
|
||||
def __getattr__(name: str) -> object:
|
||||
import warnings
|
||||
|
||||
if name == "BaseCommand":
|
||||
from .core import _BaseCommand
|
||||
|
||||
warnings.warn(
|
||||
"'BaseCommand' is deprecated and will be removed in Click 9.0. Use"
|
||||
" 'Command' instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return _BaseCommand
|
||||
|
||||
if name == "MultiCommand":
|
||||
from .core import _MultiCommand
|
||||
|
||||
warnings.warn(
|
||||
"'MultiCommand' is deprecated and will be removed in Click 9.0. Use"
|
||||
" 'Group' instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return _MultiCommand
|
||||
|
||||
if name == "OptionParser":
|
||||
from .parser import _OptionParser
|
||||
|
||||
warnings.warn(
|
||||
"'OptionParser' is deprecated and will be removed in Click 9.0. The"
|
||||
" old parser is available in 'optparse'.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return _OptionParser
|
||||
|
||||
if name == "__version__":
|
||||
import importlib.metadata
|
||||
import warnings
|
||||
|
||||
warnings.warn(
|
||||
"The '__version__' attribute is deprecated and will be removed in"
|
||||
" Click 9.1. Use feature detection or"
|
||||
" 'importlib.metadata.version(\"click\")' instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return importlib.metadata.version("click")
|
||||
|
||||
raise AttributeError(name)
|
||||
626
venv/Lib/site-packages/click/_compat.py
Normal file
626
venv/Lib/site-packages/click/_compat.py
Normal file
@@ -0,0 +1,626 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import codecs
|
||||
import collections.abc as cabc
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import typing as t
|
||||
from types import TracebackType
|
||||
from weakref import WeakKeyDictionary
|
||||
|
||||
CYGWIN = sys.platform.startswith("cygwin")
|
||||
WIN = sys.platform.startswith("win")
|
||||
auto_wrap_for_ansi: t.Callable[[t.TextIO], t.TextIO] | None = None
|
||||
_ansi_re = re.compile(r"\033\[[;?0-9]*[a-zA-Z]")
|
||||
|
||||
|
||||
def _make_text_stream(
|
||||
stream: t.BinaryIO,
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
force_readable: bool = False,
|
||||
force_writable: bool = False,
|
||||
) -> t.TextIO:
|
||||
if encoding is None:
|
||||
encoding = get_best_encoding(stream)
|
||||
if errors is None:
|
||||
errors = "replace"
|
||||
return _NonClosingTextIOWrapper(
|
||||
stream,
|
||||
encoding,
|
||||
errors,
|
||||
line_buffering=True,
|
||||
force_readable=force_readable,
|
||||
force_writable=force_writable,
|
||||
)
|
||||
|
||||
|
||||
def is_ascii_encoding(encoding: str) -> bool:
|
||||
"""Checks if a given encoding is ascii."""
|
||||
try:
|
||||
return codecs.lookup(encoding).name == "ascii"
|
||||
except LookupError:
|
||||
return False
|
||||
|
||||
|
||||
def get_best_encoding(stream: t.IO[t.Any]) -> str:
|
||||
"""Returns the default stream encoding if not found."""
|
||||
rv = getattr(stream, "encoding", None) or sys.getdefaultencoding()
|
||||
if is_ascii_encoding(rv):
|
||||
return "utf-8"
|
||||
return rv
|
||||
|
||||
|
||||
class _NonClosingTextIOWrapper(io.TextIOWrapper):
|
||||
def __init__(
|
||||
self,
|
||||
stream: t.BinaryIO,
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
force_readable: bool = False,
|
||||
force_writable: bool = False,
|
||||
**extra: t.Any,
|
||||
) -> None:
|
||||
self._stream = stream = t.cast(
|
||||
t.BinaryIO, _FixupStream(stream, force_readable, force_writable)
|
||||
)
|
||||
super().__init__(stream, encoding, errors, **extra)
|
||||
|
||||
def __del__(self) -> None:
|
||||
try:
|
||||
self.detach()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def isatty(self) -> bool:
|
||||
# https://bitbucket.org/pypy/pypy/issue/1803
|
||||
return self._stream.isatty()
|
||||
|
||||
|
||||
class _FixupStream:
|
||||
"""The new io interface needs more from streams than streams
|
||||
traditionally implement. As such, this fix-up code is necessary in
|
||||
some circumstances.
|
||||
|
||||
The forcing of readable and writable flags are there because some tools
|
||||
put badly patched objects on sys (one such offender are certain version
|
||||
of jupyter notebook).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
stream: t.BinaryIO,
|
||||
force_readable: bool = False,
|
||||
force_writable: bool = False,
|
||||
):
|
||||
self._stream = stream
|
||||
self._force_readable = force_readable
|
||||
self._force_writable = force_writable
|
||||
|
||||
def __getattr__(self, name: str) -> t.Any:
|
||||
return getattr(self._stream, name)
|
||||
|
||||
def read1(self, size: int) -> bytes:
|
||||
f = getattr(self._stream, "read1", None)
|
||||
|
||||
if f is not None:
|
||||
return t.cast(bytes, f(size))
|
||||
|
||||
return self._stream.read(size)
|
||||
|
||||
def readable(self) -> bool:
|
||||
if self._force_readable:
|
||||
return True
|
||||
x = getattr(self._stream, "readable", None)
|
||||
if x is not None:
|
||||
return t.cast(bool, x())
|
||||
try:
|
||||
self._stream.read(0)
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
def writable(self) -> bool:
|
||||
if self._force_writable:
|
||||
return True
|
||||
x = getattr(self._stream, "writable", None)
|
||||
if x is not None:
|
||||
return t.cast(bool, x())
|
||||
try:
|
||||
self._stream.write(b"")
|
||||
except Exception:
|
||||
try:
|
||||
self._stream.write(b"")
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
def seekable(self) -> bool:
|
||||
x = getattr(self._stream, "seekable", None)
|
||||
if x is not None:
|
||||
return t.cast(bool, x())
|
||||
try:
|
||||
self._stream.seek(self._stream.tell())
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _is_binary_reader(stream: t.IO[t.Any], default: bool = False) -> bool:
|
||||
try:
|
||||
return isinstance(stream.read(0), bytes)
|
||||
except Exception:
|
||||
return default
|
||||
# This happens in some cases where the stream was already
|
||||
# closed. In this case, we assume the default.
|
||||
|
||||
|
||||
def _is_binary_writer(stream: t.IO[t.Any], default: bool = False) -> bool:
|
||||
try:
|
||||
stream.write(b"")
|
||||
except Exception:
|
||||
try:
|
||||
stream.write("")
|
||||
return False
|
||||
except Exception:
|
||||
pass
|
||||
return default
|
||||
return True
|
||||
|
||||
|
||||
def _find_binary_reader(stream: t.IO[t.Any]) -> t.BinaryIO | None:
|
||||
# We need to figure out if the given stream is already binary.
|
||||
# This can happen because the official docs recommend detaching
|
||||
# the streams to get binary streams. Some code might do this, so
|
||||
# we need to deal with this case explicitly.
|
||||
if _is_binary_reader(stream, False):
|
||||
return t.cast(t.BinaryIO, stream)
|
||||
|
||||
buf = getattr(stream, "buffer", None)
|
||||
|
||||
# Same situation here; this time we assume that the buffer is
|
||||
# actually binary in case it's closed.
|
||||
if buf is not None and _is_binary_reader(buf, True):
|
||||
return t.cast(t.BinaryIO, buf)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _find_binary_writer(stream: t.IO[t.Any]) -> t.BinaryIO | None:
|
||||
# We need to figure out if the given stream is already binary.
|
||||
# This can happen because the official docs recommend detaching
|
||||
# the streams to get binary streams. Some code might do this, so
|
||||
# we need to deal with this case explicitly.
|
||||
if _is_binary_writer(stream, False):
|
||||
return t.cast(t.BinaryIO, stream)
|
||||
|
||||
buf = getattr(stream, "buffer", None)
|
||||
|
||||
# Same situation here; this time we assume that the buffer is
|
||||
# actually binary in case it's closed.
|
||||
if buf is not None and _is_binary_writer(buf, True):
|
||||
return t.cast(t.BinaryIO, buf)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _stream_is_misconfigured(stream: t.TextIO) -> bool:
|
||||
"""A stream is misconfigured if its encoding is ASCII."""
|
||||
# If the stream does not have an encoding set, we assume it's set
|
||||
# to ASCII. This appears to happen in certain unittest
|
||||
# environments. It's not quite clear what the correct behavior is
|
||||
# but this at least will force Click to recover somehow.
|
||||
return is_ascii_encoding(getattr(stream, "encoding", None) or "ascii")
|
||||
|
||||
|
||||
def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: str | None) -> bool:
|
||||
"""A stream attribute is compatible if it is equal to the
|
||||
desired value or the desired value is unset and the attribute
|
||||
has a value.
|
||||
"""
|
||||
stream_value = getattr(stream, attr, None)
|
||||
return stream_value == value or (value is None and stream_value is not None)
|
||||
|
||||
|
||||
def _is_compatible_text_stream(
|
||||
stream: t.TextIO, encoding: str | None, errors: str | None
|
||||
) -> bool:
|
||||
"""Check if a stream's encoding and errors attributes are
|
||||
compatible with the desired values.
|
||||
"""
|
||||
return _is_compat_stream_attr(
|
||||
stream, "encoding", encoding
|
||||
) and _is_compat_stream_attr(stream, "errors", errors)
|
||||
|
||||
|
||||
def _force_correct_text_stream(
|
||||
text_stream: t.IO[t.Any],
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
is_binary: t.Callable[[t.IO[t.Any], bool], bool],
|
||||
find_binary: t.Callable[[t.IO[t.Any]], t.BinaryIO | None],
|
||||
force_readable: bool = False,
|
||||
force_writable: bool = False,
|
||||
) -> t.TextIO:
|
||||
if is_binary(text_stream, False):
|
||||
binary_reader = t.cast(t.BinaryIO, text_stream)
|
||||
else:
|
||||
text_stream = t.cast(t.TextIO, text_stream)
|
||||
# If the stream looks compatible, and won't default to a
|
||||
# misconfigured ascii encoding, return it as-is.
|
||||
if _is_compatible_text_stream(text_stream, encoding, errors) and not (
|
||||
encoding is None and _stream_is_misconfigured(text_stream)
|
||||
):
|
||||
return text_stream
|
||||
|
||||
# Otherwise, get the underlying binary reader.
|
||||
possible_binary_reader = find_binary(text_stream)
|
||||
|
||||
# If that's not possible, silently use the original reader
|
||||
# and get mojibake instead of exceptions.
|
||||
if possible_binary_reader is None:
|
||||
return text_stream
|
||||
|
||||
binary_reader = possible_binary_reader
|
||||
|
||||
# Default errors to replace instead of strict in order to get
|
||||
# something that works.
|
||||
if errors is None:
|
||||
errors = "replace"
|
||||
|
||||
# Wrap the binary stream in a text stream with the correct
|
||||
# encoding parameters.
|
||||
return _make_text_stream(
|
||||
binary_reader,
|
||||
encoding,
|
||||
errors,
|
||||
force_readable=force_readable,
|
||||
force_writable=force_writable,
|
||||
)
|
||||
|
||||
|
||||
def _force_correct_text_reader(
|
||||
text_reader: t.IO[t.Any],
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
force_readable: bool = False,
|
||||
) -> t.TextIO:
|
||||
return _force_correct_text_stream(
|
||||
text_reader,
|
||||
encoding,
|
||||
errors,
|
||||
_is_binary_reader,
|
||||
_find_binary_reader,
|
||||
force_readable=force_readable,
|
||||
)
|
||||
|
||||
|
||||
def _force_correct_text_writer(
|
||||
text_writer: t.IO[t.Any],
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
force_writable: bool = False,
|
||||
) -> t.TextIO:
|
||||
return _force_correct_text_stream(
|
||||
text_writer,
|
||||
encoding,
|
||||
errors,
|
||||
_is_binary_writer,
|
||||
_find_binary_writer,
|
||||
force_writable=force_writable,
|
||||
)
|
||||
|
||||
|
||||
def get_binary_stdin() -> t.BinaryIO:
|
||||
reader = _find_binary_reader(sys.stdin)
|
||||
if reader is None:
|
||||
raise RuntimeError("Was not able to determine binary stream for sys.stdin.")
|
||||
return reader
|
||||
|
||||
|
||||
def get_binary_stdout() -> t.BinaryIO:
|
||||
writer = _find_binary_writer(sys.stdout)
|
||||
if writer is None:
|
||||
raise RuntimeError("Was not able to determine binary stream for sys.stdout.")
|
||||
return writer
|
||||
|
||||
|
||||
def get_binary_stderr() -> t.BinaryIO:
|
||||
writer = _find_binary_writer(sys.stderr)
|
||||
if writer is None:
|
||||
raise RuntimeError("Was not able to determine binary stream for sys.stderr.")
|
||||
return writer
|
||||
|
||||
|
||||
def get_text_stdin(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
|
||||
rv = _get_windows_console_stream(sys.stdin, encoding, errors)
|
||||
if rv is not None:
|
||||
return rv
|
||||
return _force_correct_text_reader(sys.stdin, encoding, errors, force_readable=True)
|
||||
|
||||
|
||||
def get_text_stdout(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
|
||||
rv = _get_windows_console_stream(sys.stdout, encoding, errors)
|
||||
if rv is not None:
|
||||
return rv
|
||||
return _force_correct_text_writer(sys.stdout, encoding, errors, force_writable=True)
|
||||
|
||||
|
||||
def get_text_stderr(encoding: str | None = None, errors: str | None = None) -> t.TextIO:
|
||||
rv = _get_windows_console_stream(sys.stderr, encoding, errors)
|
||||
if rv is not None:
|
||||
return rv
|
||||
return _force_correct_text_writer(sys.stderr, encoding, errors, force_writable=True)
|
||||
|
||||
|
||||
def _wrap_io_open(
|
||||
file: str | os.PathLike[str] | int,
|
||||
mode: str,
|
||||
encoding: str | None,
|
||||
errors: str | None,
|
||||
) -> t.IO[t.Any]:
|
||||
"""Handles not passing ``encoding`` and ``errors`` in binary mode."""
|
||||
if "b" in mode:
|
||||
return open(file, mode)
|
||||
|
||||
return open(file, mode, encoding=encoding, errors=errors)
|
||||
|
||||
|
||||
def open_stream(
|
||||
filename: str | os.PathLike[str],
|
||||
mode: str = "r",
|
||||
encoding: str | None = None,
|
||||
errors: str | None = "strict",
|
||||
atomic: bool = False,
|
||||
) -> tuple[t.IO[t.Any], bool]:
|
||||
binary = "b" in mode
|
||||
filename = os.fspath(filename)
|
||||
|
||||
# Standard streams first. These are simple because they ignore the
|
||||
# atomic flag. Use fsdecode to handle Path("-").
|
||||
if os.fsdecode(filename) == "-":
|
||||
if any(m in mode for m in ["w", "a", "x"]):
|
||||
if binary:
|
||||
return get_binary_stdout(), False
|
||||
return get_text_stdout(encoding=encoding, errors=errors), False
|
||||
if binary:
|
||||
return get_binary_stdin(), False
|
||||
return get_text_stdin(encoding=encoding, errors=errors), False
|
||||
|
||||
# Non-atomic writes directly go out through the regular open functions.
|
||||
if not atomic:
|
||||
return _wrap_io_open(filename, mode, encoding, errors), True
|
||||
|
||||
# Some usability stuff for atomic writes
|
||||
if "a" in mode:
|
||||
raise ValueError(
|
||||
"Appending to an existing file is not supported, because that"
|
||||
" would involve an expensive `copy`-operation to a temporary"
|
||||
" file. Open the file in normal `w`-mode and copy explicitly"
|
||||
" if that's what you're after."
|
||||
)
|
||||
if "x" in mode:
|
||||
raise ValueError("Use the `overwrite`-parameter instead.")
|
||||
if "w" not in mode:
|
||||
raise ValueError("Atomic writes only make sense with `w`-mode.")
|
||||
|
||||
# Atomic writes are more complicated. They work by opening a file
|
||||
# as a proxy in the same folder and then using the fdopen
|
||||
# functionality to wrap it in a Python file. Then we wrap it in an
|
||||
# atomic file that moves the file over on close.
|
||||
import errno
|
||||
import random
|
||||
|
||||
try:
|
||||
perm: int | None = os.stat(filename).st_mode
|
||||
except OSError:
|
||||
perm = None
|
||||
|
||||
flags = os.O_RDWR | os.O_CREAT | os.O_EXCL
|
||||
|
||||
if binary:
|
||||
flags |= getattr(os, "O_BINARY", 0)
|
||||
|
||||
while True:
|
||||
tmp_filename = os.path.join(
|
||||
os.path.dirname(filename),
|
||||
f".__atomic-write{random.randrange(1 << 32):08x}",
|
||||
)
|
||||
try:
|
||||
fd = os.open(tmp_filename, flags, 0o666 if perm is None else perm)
|
||||
break
|
||||
except OSError as e:
|
||||
if e.errno == errno.EEXIST or (
|
||||
os.name == "nt"
|
||||
and e.errno == errno.EACCES
|
||||
and os.path.isdir(e.filename)
|
||||
and os.access(e.filename, os.W_OK)
|
||||
):
|
||||
continue
|
||||
raise
|
||||
|
||||
if perm is not None:
|
||||
os.chmod(tmp_filename, perm) # in case perm includes bits in umask
|
||||
|
||||
f = _wrap_io_open(fd, mode, encoding, errors)
|
||||
af = _AtomicFile(f, tmp_filename, os.path.realpath(filename))
|
||||
return t.cast(t.IO[t.Any], af), True
|
||||
|
||||
|
||||
class _AtomicFile:
|
||||
def __init__(self, f: t.IO[t.Any], tmp_filename: str, real_filename: str) -> None:
|
||||
self._f = f
|
||||
self._tmp_filename = tmp_filename
|
||||
self._real_filename = real_filename
|
||||
self.closed = False
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._real_filename
|
||||
|
||||
def close(self, delete: bool = False) -> None:
|
||||
if self.closed:
|
||||
return
|
||||
self._f.close()
|
||||
os.replace(self._tmp_filename, self._real_filename)
|
||||
self.closed = True
|
||||
|
||||
def __getattr__(self, name: str) -> t.Any:
|
||||
return getattr(self._f, name)
|
||||
|
||||
def __enter__(self) -> _AtomicFile:
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exc_type: type[BaseException] | None,
|
||||
exc_value: BaseException | None,
|
||||
tb: TracebackType | None,
|
||||
) -> None:
|
||||
self.close(delete=exc_type is not None)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return repr(self._f)
|
||||
|
||||
|
||||
def strip_ansi(value: str) -> str:
|
||||
return _ansi_re.sub("", value)
|
||||
|
||||
|
||||
def _is_jupyter_kernel_output(stream: t.IO[t.Any]) -> bool:
|
||||
while isinstance(stream, (_FixupStream, _NonClosingTextIOWrapper)):
|
||||
stream = stream._stream
|
||||
|
||||
return stream.__class__.__module__.startswith("ipykernel.")
|
||||
|
||||
|
||||
def should_strip_ansi(
|
||||
stream: t.IO[t.Any] | None = None, color: bool | None = None
|
||||
) -> bool:
|
||||
if color is None:
|
||||
if stream is None:
|
||||
stream = sys.stdin
|
||||
elif hasattr(stream, "color"):
|
||||
# ._termui_impl.MaybeStripAnsi handles stripping ansi itself,
|
||||
# so we don't need to strip it here
|
||||
return False
|
||||
return not isatty(stream) and not _is_jupyter_kernel_output(stream)
|
||||
return not color
|
||||
|
||||
|
||||
# On Windows, wrap the output streams with colorama to support ANSI
|
||||
# color codes.
|
||||
# NOTE: double check is needed so mypy does not analyze this on Linux
|
||||
if sys.platform.startswith("win") and WIN:
|
||||
from ._winconsole import _get_windows_console_stream
|
||||
|
||||
def _get_argv_encoding() -> str:
|
||||
import locale
|
||||
|
||||
return locale.getpreferredencoding()
|
||||
|
||||
_ansi_stream_wrappers: cabc.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()
|
||||
|
||||
def auto_wrap_for_ansi(stream: t.TextIO, color: bool | None = None) -> t.TextIO:
|
||||
"""Support ANSI color and style codes on Windows by wrapping a
|
||||
stream with colorama.
|
||||
"""
|
||||
try:
|
||||
cached = _ansi_stream_wrappers.get(stream)
|
||||
except Exception:
|
||||
cached = None
|
||||
|
||||
if cached is not None:
|
||||
return cached
|
||||
|
||||
import colorama
|
||||
|
||||
strip = should_strip_ansi(stream, color)
|
||||
ansi_wrapper = colorama.AnsiToWin32(stream, strip=strip)
|
||||
rv = t.cast(t.TextIO, ansi_wrapper.stream)
|
||||
_write = rv.write
|
||||
|
||||
def _safe_write(s: str) -> int:
|
||||
try:
|
||||
return _write(s)
|
||||
except BaseException:
|
||||
ansi_wrapper.reset_all()
|
||||
raise
|
||||
|
||||
rv.write = _safe_write # type: ignore[method-assign]
|
||||
|
||||
try:
|
||||
_ansi_stream_wrappers[stream] = rv
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return rv
|
||||
|
||||
else:
|
||||
|
||||
def _get_argv_encoding() -> str:
|
||||
return getattr(sys.stdin, "encoding", None) or sys.getfilesystemencoding()
|
||||
|
||||
def _get_windows_console_stream(
|
||||
f: t.TextIO, encoding: str | None, errors: str | None
|
||||
) -> t.TextIO | None:
|
||||
return None
|
||||
|
||||
|
||||
def term_len(x: str) -> int:
|
||||
return len(strip_ansi(x))
|
||||
|
||||
|
||||
def isatty(stream: t.IO[t.Any]) -> bool:
|
||||
try:
|
||||
return stream.isatty()
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _make_cached_stream_func(
|
||||
src_func: t.Callable[[], t.TextIO | None],
|
||||
wrapper_func: t.Callable[[], t.TextIO],
|
||||
) -> t.Callable[[], t.TextIO | None]:
|
||||
cache: cabc.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()
|
||||
|
||||
def func() -> t.TextIO | None:
|
||||
stream = src_func()
|
||||
|
||||
if stream is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
rv = cache.get(stream)
|
||||
except Exception:
|
||||
rv = None
|
||||
if rv is not None:
|
||||
return rv
|
||||
rv = wrapper_func()
|
||||
try:
|
||||
cache[stream] = rv
|
||||
except Exception:
|
||||
pass
|
||||
return rv
|
||||
|
||||
return func
|
||||
|
||||
|
||||
_default_text_stdin = _make_cached_stream_func(lambda: sys.stdin, get_text_stdin)
|
||||
_default_text_stdout = _make_cached_stream_func(lambda: sys.stdout, get_text_stdout)
|
||||
_default_text_stderr = _make_cached_stream_func(lambda: sys.stderr, get_text_stderr)
|
||||
|
||||
|
||||
binary_streams: cabc.Mapping[str, t.Callable[[], t.BinaryIO]] = {
|
||||
"stdin": get_binary_stdin,
|
||||
"stdout": get_binary_stdout,
|
||||
"stderr": get_binary_stderr,
|
||||
}
|
||||
|
||||
text_streams: cabc.Mapping[str, t.Callable[[str | None, str | None], t.TextIO]] = {
|
||||
"stdin": get_text_stdin,
|
||||
"stdout": get_text_stdout,
|
||||
"stderr": get_text_stderr,
|
||||
}
|
||||
945
venv/Lib/site-packages/click/_termui_impl.py
Normal file
945
venv/Lib/site-packages/click/_termui_impl.py
Normal file
@@ -0,0 +1,945 @@
|
||||
"""
|
||||
This module contains implementations for the termui module. To keep the
|
||||
import time of Click down, some infrequently used functionality is
|
||||
placed in this module and only imported as needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import collections.abc as cabc
|
||||
import contextlib
|
||||
import io
|
||||
import math
|
||||
import os
|
||||
import shlex
|
||||
import sys
|
||||
import time
|
||||
import typing as t
|
||||
from gettext import gettext as _
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
from types import TracebackType
|
||||
|
||||
from ._compat import _default_text_stdout
|
||||
from ._compat import CYGWIN
|
||||
from ._compat import get_best_encoding
|
||||
from ._compat import isatty
|
||||
from ._compat import strip_ansi
|
||||
from ._compat import term_len
|
||||
from ._compat import WIN
|
||||
from .exceptions import ClickException
|
||||
from .utils import echo
|
||||
from .utils import KeepOpenFile
|
||||
|
||||
V = t.TypeVar("V")
|
||||
|
||||
|
||||
class _BufferedTextPagerStream(t.Protocol):
|
||||
buffer: t.BinaryIO
|
||||
|
||||
|
||||
def _has_binary_buffer(
|
||||
stream: t.BinaryIO | t.TextIO,
|
||||
) -> t.TypeGuard[_BufferedTextPagerStream]:
|
||||
# TextIO is wider than TextIOWrapper; text-only streams such as StringIO
|
||||
# are valid TextIO values but do not expose a binary buffer to wrap.
|
||||
return getattr(stream, "buffer", None) is not None
|
||||
|
||||
|
||||
if os.name == "nt":
|
||||
BEFORE_BAR = "\r"
|
||||
AFTER_BAR = "\n"
|
||||
else:
|
||||
BEFORE_BAR = "\r\033[?25l"
|
||||
AFTER_BAR = "\033[?25h\n"
|
||||
|
||||
|
||||
class ProgressBar(t.Generic[V]):
|
||||
def __init__(
|
||||
self,
|
||||
iterable: cabc.Iterable[V] | None,
|
||||
length: int | None = None,
|
||||
fill_char: str = "#",
|
||||
empty_char: str = " ",
|
||||
bar_template: str = "%(bar)s",
|
||||
info_sep: str = " ",
|
||||
hidden: bool = False,
|
||||
show_eta: bool = True,
|
||||
show_percent: bool | None = None,
|
||||
show_pos: bool = False,
|
||||
item_show_func: t.Callable[[V | None], str | None] | None = None,
|
||||
label: str | None = None,
|
||||
file: t.TextIO | None = None,
|
||||
color: bool | None = None,
|
||||
update_min_steps: int = 1,
|
||||
width: int = 30,
|
||||
) -> None:
|
||||
self.fill_char = fill_char
|
||||
self.empty_char = empty_char
|
||||
self.bar_template = bar_template
|
||||
self.info_sep = info_sep
|
||||
self.hidden = hidden
|
||||
self.show_eta = show_eta
|
||||
self.show_percent = show_percent
|
||||
self.show_pos = show_pos
|
||||
self.item_show_func = item_show_func
|
||||
self.label: str = label or ""
|
||||
|
||||
if file is None:
|
||||
file = _default_text_stdout()
|
||||
|
||||
# There are no standard streams attached to write to. For example,
|
||||
# pythonw on Windows.
|
||||
if file is None:
|
||||
file = StringIO()
|
||||
|
||||
self.file = file
|
||||
self.color = color
|
||||
self.update_min_steps = update_min_steps
|
||||
self._completed_intervals = 0
|
||||
self.width: int = width
|
||||
self.autowidth: bool = width == 0
|
||||
|
||||
if length is None:
|
||||
from operator import length_hint
|
||||
|
||||
length = length_hint(iterable, -1)
|
||||
|
||||
if length == -1:
|
||||
length = None
|
||||
if iterable is None:
|
||||
if length is None:
|
||||
raise TypeError("iterable or length is required")
|
||||
iterable = t.cast("cabc.Iterable[V]", range(length))
|
||||
self.iter: cabc.Iterable[V] = iter(iterable)
|
||||
self.length = length
|
||||
self.pos: int = 0
|
||||
self.avg: list[float] = []
|
||||
self.last_eta: float
|
||||
self.start: float
|
||||
self.start = self.last_eta = time.time()
|
||||
self.eta_known: bool = False
|
||||
self.finished: bool = False
|
||||
self.max_width: int | None = None
|
||||
self.entered: bool = False
|
||||
self.current_item: V | None = None
|
||||
self._is_atty = isatty(self.file)
|
||||
self._last_line: str | None = None
|
||||
|
||||
def __enter__(self) -> ProgressBar[V]:
|
||||
self.entered = True
|
||||
self.render_progress()
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exc_type: type[BaseException] | None,
|
||||
exc_value: BaseException | None,
|
||||
tb: TracebackType | None,
|
||||
) -> None:
|
||||
self.render_finish()
|
||||
|
||||
def __iter__(self) -> cabc.Iterator[V]:
|
||||
if not self.entered:
|
||||
raise RuntimeError("You need to use progress bars in a with block.")
|
||||
self.render_progress()
|
||||
return self.generator()
|
||||
|
||||
def __next__(self) -> V:
|
||||
# Iteration is defined in terms of a generator function,
|
||||
# returned by iter(self); use that to define next(). This works
|
||||
# because `self.iter` is an iterable consumed by that generator,
|
||||
# so it is re-entry safe. Calling `next(self.generator())`
|
||||
# twice works and does "what you want".
|
||||
return next(iter(self))
|
||||
|
||||
def render_finish(self) -> None:
|
||||
if self.hidden or not self._is_atty:
|
||||
return
|
||||
self.file.write(AFTER_BAR)
|
||||
self.file.flush()
|
||||
|
||||
@property
|
||||
def pct(self) -> float:
|
||||
if self.finished:
|
||||
return 1.0
|
||||
return min(self.pos / (float(self.length or 1) or 1), 1.0)
|
||||
|
||||
@property
|
||||
def time_per_iteration(self) -> float:
|
||||
if not self.avg:
|
||||
return 0.0
|
||||
return sum(self.avg) / float(len(self.avg))
|
||||
|
||||
@property
|
||||
def eta(self) -> float:
|
||||
if self.length is not None and not self.finished:
|
||||
return self.time_per_iteration * (self.length - self.pos)
|
||||
return 0.0
|
||||
|
||||
def format_eta(self) -> str:
|
||||
if self.eta_known:
|
||||
t = int(self.eta)
|
||||
seconds = t % 60
|
||||
t //= 60
|
||||
minutes = t % 60
|
||||
t //= 60
|
||||
hours = t % 24
|
||||
t //= 24
|
||||
if t > 0:
|
||||
return "{d}{day_label} {h:02}:{m:02}:{s:02}".format(
|
||||
d=t,
|
||||
day_label=_("d"),
|
||||
h=hours,
|
||||
m=minutes,
|
||||
s=seconds,
|
||||
)
|
||||
else:
|
||||
return f"{hours:02}:{minutes:02}:{seconds:02}"
|
||||
return ""
|
||||
|
||||
def format_pos(self) -> str:
|
||||
pos = str(self.pos)
|
||||
if self.length is not None:
|
||||
pos += f"/{self.length}"
|
||||
return pos
|
||||
|
||||
def format_pct(self) -> str:
|
||||
return f"{int(self.pct * 100): 4}%"[1:]
|
||||
|
||||
def format_bar(self) -> str:
|
||||
if self.length is not None:
|
||||
bar_length = int(self.pct * self.width)
|
||||
bar = self.fill_char * bar_length
|
||||
bar += self.empty_char * (self.width - bar_length)
|
||||
elif self.finished:
|
||||
bar = self.fill_char * self.width
|
||||
else:
|
||||
chars = list(self.empty_char * (self.width or 1))
|
||||
if self.time_per_iteration != 0:
|
||||
chars[
|
||||
int(
|
||||
(math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
|
||||
* self.width
|
||||
)
|
||||
] = self.fill_char
|
||||
bar = "".join(chars)
|
||||
return bar
|
||||
|
||||
def format_progress_line(self) -> str:
|
||||
show_percent = self.show_percent
|
||||
|
||||
info_bits = []
|
||||
if self.length is not None and show_percent is None:
|
||||
show_percent = not self.show_pos
|
||||
|
||||
if self.show_pos:
|
||||
info_bits.append(self.format_pos())
|
||||
if show_percent:
|
||||
info_bits.append(self.format_pct())
|
||||
if self.show_eta and self.eta_known and not self.finished:
|
||||
info_bits.append(self.format_eta())
|
||||
if self.item_show_func is not None:
|
||||
item_info = self.item_show_func(self.current_item)
|
||||
if item_info is not None:
|
||||
info_bits.append(item_info)
|
||||
|
||||
return (
|
||||
self.bar_template
|
||||
% {
|
||||
"label": self.label,
|
||||
"bar": self.format_bar(),
|
||||
"info": self.info_sep.join(info_bits),
|
||||
}
|
||||
).rstrip()
|
||||
|
||||
def render_progress(self) -> None:
|
||||
if self.hidden:
|
||||
return
|
||||
|
||||
if not self._is_atty:
|
||||
# Only output the label once if the output is not a TTY.
|
||||
if self._last_line != self.label:
|
||||
self._last_line = self.label
|
||||
echo(self.label, file=self.file, color=self.color)
|
||||
return
|
||||
|
||||
buf = []
|
||||
# Update width in case the terminal has been resized
|
||||
if self.autowidth:
|
||||
import shutil
|
||||
|
||||
old_width = self.width
|
||||
self.width = 0
|
||||
clutter_length = term_len(self.format_progress_line())
|
||||
new_width = max(0, shutil.get_terminal_size().columns - clutter_length)
|
||||
if new_width < old_width and self.max_width is not None:
|
||||
buf.append(BEFORE_BAR)
|
||||
buf.append(" " * self.max_width)
|
||||
self.max_width = new_width
|
||||
self.width = new_width
|
||||
|
||||
clear_width = self.width
|
||||
if self.max_width is not None:
|
||||
clear_width = self.max_width
|
||||
|
||||
buf.append(BEFORE_BAR)
|
||||
line = self.format_progress_line()
|
||||
line_len = term_len(line)
|
||||
if self.max_width is None or self.max_width < line_len:
|
||||
self.max_width = line_len
|
||||
|
||||
buf.append(line)
|
||||
buf.append(" " * (clear_width - line_len))
|
||||
line = "".join(buf)
|
||||
# Render the line only if it changed.
|
||||
|
||||
if line != self._last_line:
|
||||
self._last_line = line
|
||||
echo(line, file=self.file, color=self.color, nl=False)
|
||||
self.file.flush()
|
||||
|
||||
def make_step(self, n_steps: int) -> None:
|
||||
self.pos += n_steps
|
||||
if self.length is not None and self.pos >= self.length:
|
||||
self.finished = True
|
||||
|
||||
if (time.time() - self.last_eta) < 1.0:
|
||||
return
|
||||
|
||||
self.last_eta = time.time()
|
||||
|
||||
# self.avg is a rolling list of length <= 7 of steps where steps are
|
||||
# defined as time elapsed divided by the total progress through
|
||||
# self.length.
|
||||
if self.pos:
|
||||
step = (time.time() - self.start) / self.pos
|
||||
else:
|
||||
step = time.time() - self.start
|
||||
|
||||
self.avg = self.avg[-6:] + [step]
|
||||
|
||||
self.eta_known = self.length is not None
|
||||
|
||||
def update(self, n_steps: int, current_item: V | None = None) -> None:
|
||||
"""Update the progress bar by advancing a specified number of
|
||||
steps, and optionally set the ``current_item`` for this new
|
||||
position.
|
||||
|
||||
:param n_steps: Number of steps to advance.
|
||||
:param current_item: Optional item to set as ``current_item``
|
||||
for the updated position.
|
||||
|
||||
.. versionchanged:: 8.0
|
||||
Added the ``current_item`` optional parameter.
|
||||
|
||||
.. versionchanged:: 8.0
|
||||
Only render when the number of steps meets the
|
||||
``update_min_steps`` threshold.
|
||||
"""
|
||||
if current_item is not None:
|
||||
self.current_item = current_item
|
||||
|
||||
self._completed_intervals += n_steps
|
||||
|
||||
if self._completed_intervals >= self.update_min_steps:
|
||||
self.make_step(self._completed_intervals)
|
||||
self.render_progress()
|
||||
self._completed_intervals = 0
|
||||
|
||||
def finish(self) -> None:
|
||||
self.eta_known = False
|
||||
self.current_item = None
|
||||
self.finished = True
|
||||
|
||||
def generator(self) -> cabc.Iterator[V]:
|
||||
"""Return a generator which yields the items added to the bar
|
||||
during construction, and updates the progress bar *after* the
|
||||
yielded block returns.
|
||||
"""
|
||||
# WARNING: the iterator interface for `ProgressBar` relies on
|
||||
# this and only works because this is a simple generator which
|
||||
# doesn't create or manage additional state. If this function
|
||||
# changes, the impact should be evaluated both against
|
||||
# `iter(bar)` and `next(bar)`. `next()` in particular may call
|
||||
# `self.generator()` repeatedly, and this must remain safe in
|
||||
# order for that interface to work.
|
||||
if not self.entered:
|
||||
raise RuntimeError("You need to use progress bars in a with block.")
|
||||
|
||||
if not self._is_atty:
|
||||
yield from self.iter
|
||||
else:
|
||||
for rv in self.iter:
|
||||
self.current_item = rv
|
||||
|
||||
# This allows show_item_func to be updated before the
|
||||
# item is processed. Only trigger at the beginning of
|
||||
# the update interval.
|
||||
if self._completed_intervals == 0:
|
||||
self.render_progress()
|
||||
|
||||
yield rv
|
||||
self.update(1)
|
||||
|
||||
self.finish()
|
||||
self.render_progress()
|
||||
|
||||
|
||||
class MaybeStripAnsi(io.TextIOWrapper):
|
||||
def __init__(self, stream: t.IO[bytes], *, color: bool, **kwargs: t.Any):
|
||||
super().__init__(stream, **kwargs)
|
||||
self.color = color
|
||||
|
||||
def write(self, text: str) -> int:
|
||||
if not self.color:
|
||||
text = strip_ansi(text)
|
||||
return super().write(text)
|
||||
|
||||
|
||||
def _pager_contextmanager(
|
||||
color: bool | None = None,
|
||||
) -> t.ContextManager[tuple[t.BinaryIO | t.TextIO, str, bool]]:
|
||||
"""Decide what method to use for paging through text."""
|
||||
stdout = _default_text_stdout()
|
||||
|
||||
# There are no standard streams attached to write to. For example,
|
||||
# pythonw on Windows.
|
||||
if stdout is None:
|
||||
stdout = StringIO()
|
||||
|
||||
if not isatty(sys.stdin) or not isatty(stdout):
|
||||
return _nullpager(stdout, color)
|
||||
|
||||
# Split using POSIX mode (the default) so that quote characters are
|
||||
# stripped from tokens and quoted Windows paths are preserved.
|
||||
# Non-POSIX mode retains quotes in tokens, and wrapping tokens
|
||||
# with shlex.quote re-introduces quoting issues on Windows.
|
||||
pager_cmd_parts = shlex.split(os.environ.get("PAGER", ""))
|
||||
if pager_cmd_parts:
|
||||
if WIN:
|
||||
return _tempfilepager(pager_cmd_parts, color)
|
||||
return _pipepager(pager_cmd_parts, color)
|
||||
|
||||
if os.environ.get("TERM") in ("dumb", "emacs"):
|
||||
return _nullpager(stdout, color)
|
||||
if WIN or sys.platform.startswith("os2"):
|
||||
return _tempfilepager(["more"], color)
|
||||
return _pipepager(["less"], color)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def get_pager_file(color: bool | None = None) -> t.Generator[t.TextIO, None, None]:
|
||||
"""Context manager.
|
||||
|
||||
Yields a writable file-like object which can be used as an output pager.
|
||||
|
||||
.. versionadded:: 8.4.0
|
||||
|
||||
:param color: controls if the pager supports ANSI colors or not. The
|
||||
default is autodetection.
|
||||
"""
|
||||
with _pager_contextmanager(color=color) as (stream, encoding, color):
|
||||
# Split streams by capabilities rather than the abstract TextIO /
|
||||
# BinaryIO annotations: buffered text streams can be unwrapped to bytes,
|
||||
# while other streams are yielded as-is.
|
||||
wrapper: MaybeStripAnsi | None = None
|
||||
if _has_binary_buffer(stream):
|
||||
# Text stream backed by a binary buffer.
|
||||
wrapper = MaybeStripAnsi(stream.buffer, color=color, encoding=encoding)
|
||||
stream = wrapper
|
||||
try:
|
||||
# Narrow the BinaryIO | TextIO union that _pager_contextmanager
|
||||
# yields; the caller writes text to the pager.
|
||||
yield t.cast(t.TextIO, stream)
|
||||
finally:
|
||||
try:
|
||||
stream.flush()
|
||||
finally:
|
||||
# Hand the binary buffer back to the pager that produced it
|
||||
# rather than letting this TextIOWrapper close it on garbage
|
||||
# collection. The pager owns the buffer's lifecycle: subprocess
|
||||
# pipes and temp files are closed by their own helpers, while a
|
||||
# borrowed stdout must stay open for the caller. detach() runs
|
||||
# even if flush() raised, so the buffer is never closed here.
|
||||
if wrapper is not None:
|
||||
wrapper.detach()
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _pipepager(
|
||||
cmd_parts: list[str], color: bool | None = None
|
||||
) -> t.Iterator[tuple[t.BinaryIO | t.TextIO, str, bool]]:
|
||||
"""Page through text by feeding it to another program.
|
||||
|
||||
Invokes the pager via :class:`subprocess.Popen` with an ``argv`` list
|
||||
produced by :func:`shlex.split`. The command is resolved to an absolute
|
||||
path with :func:`shutil.which` as recommended by the
|
||||
:mod:`subprocess` docs for Windows compatibility.
|
||||
|
||||
Invoking a pager through this might support colors: if piping to
|
||||
``less`` and the user hasn't decided on colors, ``LESS=-R`` is set
|
||||
automatically.
|
||||
"""
|
||||
# Split the command into the invoked CLI and its parameters.
|
||||
if not cmd_parts:
|
||||
# No usable pager: fall back to stdout through _nullpager so it gets the
|
||||
# same borrowed-stream handling and the caller's stream is not closed.
|
||||
stdout = _default_text_stdout() or StringIO()
|
||||
with _nullpager(stdout, color) as rv:
|
||||
yield rv
|
||||
return
|
||||
|
||||
import shutil
|
||||
|
||||
cmd = cmd_parts[0]
|
||||
cmd_params = cmd_parts[1:]
|
||||
|
||||
cmd_filepath = shutil.which(cmd)
|
||||
if not cmd_filepath:
|
||||
# No usable pager: fall back to stdout through _nullpager so it gets the
|
||||
# same borrowed-stream handling and the caller's stream is not closed.
|
||||
stdout = _default_text_stdout() or StringIO()
|
||||
with _nullpager(stdout, color) as rv:
|
||||
yield rv
|
||||
return
|
||||
|
||||
# Produces a normalized absolute path string.
|
||||
# multi-call binaries such as busybox derive their identity from the symlink
|
||||
# less -> busybox. resolve() causes them to misbehave. (eg. less becomes busybox)
|
||||
cmd_path = Path(cmd_filepath).absolute()
|
||||
cmd_name = cmd_path.name
|
||||
|
||||
import subprocess
|
||||
|
||||
# Make a local copy of the environment to not affect the global one.
|
||||
env = dict(os.environ)
|
||||
|
||||
# If we're piping to less and the user hasn't decided on colors, we enable
|
||||
# them by default we find the -R flag in the command line arguments.
|
||||
if color is None and cmd_name == "less":
|
||||
less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_params)}"
|
||||
if not less_flags:
|
||||
env["LESS"] = "-R"
|
||||
color = True
|
||||
elif "r" in less_flags or "R" in less_flags:
|
||||
color = True
|
||||
|
||||
if color is None:
|
||||
color = False
|
||||
|
||||
c = subprocess.Popen(
|
||||
[str(cmd_path)] + cmd_params,
|
||||
shell=False,
|
||||
stdin=subprocess.PIPE,
|
||||
env=env,
|
||||
errors="replace",
|
||||
text=True,
|
||||
)
|
||||
stdin = t.cast(t.BinaryIO, c.stdin)
|
||||
encoding = get_best_encoding(stdin)
|
||||
try:
|
||||
yield stdin, encoding, color
|
||||
except BrokenPipeError:
|
||||
# In case the pager exited unexpectedly, ignore the broken pipe error.
|
||||
pass
|
||||
except Exception as e:
|
||||
# In case there is an exception we want to close the pager immediately
|
||||
# and let the caller handle it.
|
||||
# Otherwise the pager will keep running, and the user may not notice
|
||||
# the error message, or worse yet it may leave the terminal in a broken state.
|
||||
c.terminate()
|
||||
raise e
|
||||
finally:
|
||||
# We must close stdin and wait for the pager to exit before we continue
|
||||
try:
|
||||
stdin.close()
|
||||
# Close implies flush, so it might throw a BrokenPipeError if the pager
|
||||
# process exited already.
|
||||
except BrokenPipeError:
|
||||
pass
|
||||
|
||||
# Less doesn't respect ^C, but catches it for its own UI purposes (aborting
|
||||
# search or other commands inside less).
|
||||
#
|
||||
# That means when the user hits ^C, the parent process (click) terminates,
|
||||
# but less is still alive, paging the output and messing up the terminal.
|
||||
#
|
||||
# If the user wants to make the pager exit on ^C, they should set
|
||||
# `LESS='-K'`. It's not our decision to make.
|
||||
while True:
|
||||
try:
|
||||
c.wait()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
else:
|
||||
break
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _tempfilepager(
|
||||
cmd_parts: list[str], color: bool | None = None
|
||||
) -> t.Iterator[tuple[t.BinaryIO | t.TextIO, str, bool]]:
|
||||
"""Page through text by invoking a program on a temporary file.
|
||||
|
||||
Used as the primary pager strategy on Windows (where piping to
|
||||
``more`` adds spurious ``\\r\\n``), and as a fallback on other
|
||||
platforms. The command is resolved to an absolute path with
|
||||
:func:`shutil.which`.
|
||||
"""
|
||||
# Split the command into the invoked CLI and its parameters.
|
||||
if not cmd_parts:
|
||||
# No usable pager: fall back to stdout through _nullpager so it gets the
|
||||
# same borrowed-stream handling and the caller's stream is not closed.
|
||||
stdout = _default_text_stdout() or StringIO()
|
||||
with _nullpager(stdout, color) as rv:
|
||||
yield rv
|
||||
return
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
cmd = cmd_parts[0]
|
||||
|
||||
cmd_filepath = shutil.which(cmd)
|
||||
if not cmd_filepath:
|
||||
# No usable pager: fall back to stdout through _nullpager so it gets the
|
||||
# same borrowed-stream handling and the caller's stream is not closed.
|
||||
stdout = _default_text_stdout() or StringIO()
|
||||
with _nullpager(stdout, color) as rv:
|
||||
yield rv
|
||||
return
|
||||
|
||||
# Produces a normalized absolute path string.
|
||||
# multi-call binaries such as busybox derive their identity from the symlink
|
||||
# less -> busybox. resolve() causes them to misbehave. (eg. less becomes busybox)
|
||||
cmd_path = Path(cmd_filepath).absolute()
|
||||
|
||||
import tempfile
|
||||
|
||||
encoding = get_best_encoding(sys.stdout)
|
||||
if color is None:
|
||||
color = False
|
||||
# On Windows, NamedTemporaryFile cannot be opened by another process
|
||||
# while Python still has it open, so we use delete=False and clean up manually
|
||||
# rather than using a contextmanager here.
|
||||
f = tempfile.NamedTemporaryFile(mode="wb", delete=False)
|
||||
try:
|
||||
yield t.cast(t.BinaryIO, f), encoding, color
|
||||
f.flush()
|
||||
f.close()
|
||||
subprocess.call([str(cmd_path), f.name])
|
||||
finally:
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _nullpager(
|
||||
stream: t.TextIO, color: bool | None = None
|
||||
) -> t.Iterator[tuple[t.TextIO, str, bool]]:
|
||||
"""Simply print unformatted text. This is the ultimate fallback. Don't close the
|
||||
output stream in this case, since it's coming from elsewhere rather than our
|
||||
internal helpers.
|
||||
|
||||
The stream is wrapped in :class:`~click.utils.KeepOpenFile` so that, as a
|
||||
borrowed stream, it is not closed by a ``with`` block. The wrapper that
|
||||
:func:`get_pager_file` builds around it is detached rather than closed.
|
||||
"""
|
||||
encoding = get_best_encoding(stream)
|
||||
|
||||
if color is None:
|
||||
color = False
|
||||
|
||||
yield KeepOpenFile(stream), encoding, color # type: ignore[misc]
|
||||
|
||||
|
||||
class Editor:
|
||||
def __init__(
|
||||
self,
|
||||
editor: str | None = None,
|
||||
env: cabc.Mapping[str, str] | None = None,
|
||||
require_save: bool = True,
|
||||
extension: str = ".txt",
|
||||
) -> None:
|
||||
self.editor = editor
|
||||
self.env = env
|
||||
self.require_save = require_save
|
||||
self.extension = extension
|
||||
|
||||
def get_editor(self) -> str:
|
||||
if self.editor is not None:
|
||||
return self.editor
|
||||
for key in "VISUAL", "EDITOR":
|
||||
rv = os.environ.get(key)
|
||||
if rv:
|
||||
return rv
|
||||
if WIN:
|
||||
return "notepad"
|
||||
|
||||
from shutil import which
|
||||
|
||||
for editor in "sensible-editor", "vim", "nano":
|
||||
if which(editor) is not None:
|
||||
return editor
|
||||
return "vi"
|
||||
|
||||
def edit_files(self, filenames: cabc.Iterable[str]) -> None:
|
||||
"""Open files in the user's editor."""
|
||||
import shlex
|
||||
import subprocess
|
||||
|
||||
editor = self.get_editor()
|
||||
environ: dict[str, str] | None = None
|
||||
|
||||
if self.env:
|
||||
environ = os.environ.copy()
|
||||
environ.update(self.env)
|
||||
|
||||
try:
|
||||
# Split in POSIX mode (the default) for the same reasons as
|
||||
# in pager(): strips quotes from tokens and preserves quoted
|
||||
# Windows paths.
|
||||
c = subprocess.Popen(
|
||||
args=shlex.split(editor) + list(filenames),
|
||||
env=environ,
|
||||
)
|
||||
exit_code = c.wait()
|
||||
if exit_code != 0:
|
||||
raise ClickException(
|
||||
_("{editor}: Editing failed").format(editor=editor)
|
||||
)
|
||||
except OSError as e:
|
||||
raise ClickException(
|
||||
_("{editor}: Editing failed: {e}").format(editor=editor, e=e)
|
||||
) from e
|
||||
|
||||
@t.overload
|
||||
def edit(self, text: bytes | bytearray) -> bytes | None: ...
|
||||
|
||||
# We cannot know whether or not the type expected is str or bytes when None
|
||||
# is passed, so str is returned as that was what was done before.
|
||||
@t.overload
|
||||
def edit(self, text: str | None) -> str | None: ...
|
||||
|
||||
def edit(self, text: str | bytes | bytearray | None) -> str | bytes | None:
|
||||
import tempfile
|
||||
|
||||
if text is None:
|
||||
data: bytes | bytearray = b""
|
||||
elif isinstance(text, (bytes, bytearray)):
|
||||
data = text
|
||||
else:
|
||||
if text and not text.endswith("\n"):
|
||||
text += "\n"
|
||||
|
||||
if WIN:
|
||||
data = text.replace("\n", "\r\n").encode("utf-8-sig")
|
||||
else:
|
||||
data = text.encode("utf-8")
|
||||
|
||||
fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)
|
||||
f: t.BinaryIO
|
||||
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
f.write(data)
|
||||
|
||||
# If the filesystem resolution is 1 second, like Mac OS
|
||||
# 10.12 Extended, or 2 seconds, like FAT32, and the editor
|
||||
# closes very fast, require_save can fail. Set the modified
|
||||
# time to be 2 seconds in the past to work around this.
|
||||
os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2))
|
||||
# Depending on the resolution, the exact value might not be
|
||||
# recorded, so get the new recorded value.
|
||||
timestamp = os.path.getmtime(name)
|
||||
|
||||
self.edit_files((name,))
|
||||
|
||||
if self.require_save and os.path.getmtime(name) == timestamp:
|
||||
return None
|
||||
|
||||
with open(name, "rb") as f:
|
||||
rv = f.read()
|
||||
|
||||
if isinstance(text, (bytes, bytearray)):
|
||||
return rv
|
||||
|
||||
return rv.decode("utf-8-sig").replace("\r\n", "\n")
|
||||
finally:
|
||||
os.unlink(name)
|
||||
|
||||
|
||||
def open_url(url: str, wait: bool = False, locate: bool = False) -> int:
|
||||
import subprocess
|
||||
|
||||
def _unquote_file(url: str) -> str:
|
||||
from urllib.parse import unquote
|
||||
|
||||
if url.startswith("file://"):
|
||||
url = unquote(url[7:])
|
||||
|
||||
return url
|
||||
|
||||
if sys.platform == "darwin":
|
||||
args = ["open"]
|
||||
if wait:
|
||||
args.append("-W")
|
||||
if locate:
|
||||
args.append("-R")
|
||||
args.append(_unquote_file(url))
|
||||
null = open("/dev/null", "w")
|
||||
try:
|
||||
return subprocess.Popen(args, stderr=null).wait()
|
||||
finally:
|
||||
null.close()
|
||||
elif WIN:
|
||||
if locate:
|
||||
url = _unquote_file(url)
|
||||
args = ["explorer", "/select,", url]
|
||||
try:
|
||||
return subprocess.call(args)
|
||||
except OSError:
|
||||
return 127
|
||||
else:
|
||||
try:
|
||||
os.startfile(url) # type: ignore[attr-defined]
|
||||
except OSError:
|
||||
return 127
|
||||
return 0
|
||||
elif CYGWIN:
|
||||
if locate:
|
||||
url = _unquote_file(url)
|
||||
args = ["cygstart", os.path.dirname(url)]
|
||||
else:
|
||||
args = ["cygstart"]
|
||||
if wait:
|
||||
args.append("-w")
|
||||
args.append(url)
|
||||
try:
|
||||
return subprocess.call(args)
|
||||
except OSError:
|
||||
# Command not found
|
||||
return 127
|
||||
|
||||
try:
|
||||
if locate:
|
||||
url = os.path.dirname(_unquote_file(url)) or "."
|
||||
else:
|
||||
url = _unquote_file(url)
|
||||
c = subprocess.Popen(["xdg-open", url])
|
||||
if wait:
|
||||
return c.wait()
|
||||
return 0
|
||||
except OSError:
|
||||
if url.startswith(("http://", "https://")) and not locate and not wait:
|
||||
import webbrowser
|
||||
|
||||
webbrowser.open(url)
|
||||
return 0
|
||||
return 1
|
||||
|
||||
|
||||
def _translate_ch_to_exc(ch: str) -> None:
|
||||
if ch == "\x03":
|
||||
raise KeyboardInterrupt()
|
||||
|
||||
if ch == "\x04" and not WIN: # Unix-like, Ctrl+D
|
||||
raise EOFError()
|
||||
|
||||
if ch == "\x1a" and WIN: # Windows, Ctrl+Z
|
||||
raise EOFError()
|
||||
|
||||
|
||||
if sys.platform == "win32":
|
||||
import msvcrt
|
||||
|
||||
@contextlib.contextmanager
|
||||
def raw_terminal() -> cabc.Iterator[int]:
|
||||
yield -1
|
||||
|
||||
def getchar(echo: bool) -> str:
|
||||
# The function `getch` will return a bytes object corresponding to
|
||||
# the pressed character. Since Windows 10 build 1803, it will also
|
||||
# return \x00 when called a second time after pressing a regular key.
|
||||
#
|
||||
# `getwch` does not share this probably-bugged behavior. Moreover, it
|
||||
# returns a Unicode object by default, which is what we want.
|
||||
#
|
||||
# Either of these functions will return \x00 or \xe0 to indicate
|
||||
# a special key, and you need to call the same function again to get
|
||||
# the "rest" of the code. The fun part is that \u00e0 is
|
||||
# "latin small letter a with grave", so if you type that on a French
|
||||
# keyboard, you _also_ get a \xe0.
|
||||
# E.g., consider the Up arrow. This returns \xe0 and then \x48. The
|
||||
# resulting Unicode string reads as "a with grave" + "capital H".
|
||||
# This is indistinguishable from when the user actually types
|
||||
# "a with grave" and then "capital H".
|
||||
#
|
||||
# When \xe0 is returned, we assume it's part of a special-key sequence
|
||||
# and call `getwch` again, but that means that when the user types
|
||||
# the \u00e0 character, `getchar` doesn't return until a second
|
||||
# character is typed.
|
||||
# The alternative is returning immediately, but that would mess up
|
||||
# cross-platform handling of arrow keys and others that start with
|
||||
# \xe0. Another option is using `getch`, but then we can't reliably
|
||||
# read non-ASCII characters, because return values of `getch` are
|
||||
# limited to the current 8-bit codepage.
|
||||
#
|
||||
# Anyway, Click doesn't claim to do this Right(tm), and using `getwch`
|
||||
# is doing the right thing in more situations than with `getch`.
|
||||
|
||||
if echo:
|
||||
func = t.cast(t.Callable[[], str], msvcrt.getwche)
|
||||
else:
|
||||
func = t.cast(t.Callable[[], str], msvcrt.getwch)
|
||||
|
||||
rv = func()
|
||||
|
||||
if rv in ("\x00", "\xe0"):
|
||||
# \x00 and \xe0 are control characters that indicate special key,
|
||||
# see above.
|
||||
rv += func()
|
||||
|
||||
_translate_ch_to_exc(rv)
|
||||
return rv
|
||||
|
||||
else:
|
||||
import termios
|
||||
import tty
|
||||
|
||||
@contextlib.contextmanager
|
||||
def raw_terminal() -> cabc.Iterator[int]:
|
||||
f: t.TextIO | None
|
||||
fd: int
|
||||
|
||||
if not isatty(sys.stdin):
|
||||
f = open("/dev/tty")
|
||||
fd = f.fileno()
|
||||
else:
|
||||
fd = sys.stdin.fileno()
|
||||
f = None
|
||||
|
||||
try:
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
|
||||
try:
|
||||
tty.setraw(fd)
|
||||
yield fd
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
sys.stdout.flush()
|
||||
|
||||
if f is not None:
|
||||
f.close()
|
||||
except termios.error:
|
||||
pass
|
||||
|
||||
def getchar(echo: bool) -> str:
|
||||
with raw_terminal() as fd:
|
||||
ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace")
|
||||
|
||||
if echo and isatty(sys.stdout):
|
||||
sys.stdout.write(ch)
|
||||
|
||||
_translate_ch_to_exc(ch)
|
||||
return ch
|
||||
188
venv/Lib/site-packages/click/_textwrap.py
Normal file
188
venv/Lib/site-packages/click/_textwrap.py
Normal file
@@ -0,0 +1,188 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import collections.abc as cabc
|
||||
import textwrap
|
||||
from contextlib import contextmanager
|
||||
|
||||
from ._compat import _ansi_re
|
||||
from ._compat import term_len
|
||||
|
||||
|
||||
def _truncate_visible(text: str, n: int) -> str:
|
||||
"""Return the longest prefix of ``text`` containing at most ``n`` visible
|
||||
characters.
|
||||
|
||||
ANSI escape sequences inside the prefix are kept intact and do not count
|
||||
toward the visible width. A cut is never placed inside an escape sequence.
|
||||
"""
|
||||
if n <= 0:
|
||||
return ""
|
||||
|
||||
visible = 0
|
||||
i = 0
|
||||
cut = 0
|
||||
end = len(text)
|
||||
while i < end:
|
||||
m = _ansi_re.match(text, i)
|
||||
if m is not None:
|
||||
i = m.end()
|
||||
continue
|
||||
visible += 1
|
||||
i += 1
|
||||
cut = i
|
||||
if visible >= n:
|
||||
break
|
||||
return text[:cut]
|
||||
|
||||
|
||||
class TextWrapper(textwrap.TextWrapper):
|
||||
"""``textwrap.TextWrapper`` variant that measures widths by visible
|
||||
character count.
|
||||
|
||||
ANSI escape sequences embedded in chunks, indents, or the placeholder are
|
||||
excluded from the width budget. Without this, styled help text (a styled
|
||||
``Usage:`` prefix, a colorized option name, ...) would be wrapped earlier
|
||||
than its visible length warrants and tokens would split mid-word.
|
||||
"""
|
||||
|
||||
def _handle_long_word(
|
||||
self,
|
||||
reversed_chunks: list[str],
|
||||
cur_line: list[str],
|
||||
cur_len: int,
|
||||
width: int,
|
||||
) -> None:
|
||||
space_left = max(width - cur_len, 1)
|
||||
|
||||
if self.break_long_words:
|
||||
last = reversed_chunks[-1]
|
||||
cut = _truncate_visible(last, space_left)
|
||||
res = last[len(cut) :]
|
||||
cur_line.append(cut)
|
||||
reversed_chunks[-1] = res
|
||||
elif not cur_line:
|
||||
cur_line.append(reversed_chunks.pop())
|
||||
|
||||
def _wrap_chunks(self, chunks: list[str]) -> list[str]:
|
||||
"""Wrap chunks counting widths in visible characters.
|
||||
|
||||
Mirrors the algorithm of :meth:`textwrap.TextWrapper._wrap_chunks`
|
||||
with every width measurement routed through
|
||||
:func:`click._compat.term_len` instead of :func:`len`, so ANSI escape
|
||||
bytes in chunks, indents, or the placeholder do not inflate the count.
|
||||
|
||||
.. seealso::
|
||||
:class:`textwrap.TextWrapper` in the Python standard library documentation:
|
||||
https://docs.python.org/3/library/textwrap.html#textwrap.TextWrapper
|
||||
|
||||
Reference implementation in CPython:
|
||||
https://github.com/python/cpython/blob/main/Lib/textwrap.py
|
||||
"""
|
||||
lines: list[str] = []
|
||||
if self.width <= 0:
|
||||
raise ValueError(f"invalid width {self.width!r} (must be > 0)")
|
||||
if self.max_lines is not None:
|
||||
if self.max_lines > 1:
|
||||
indent = self.subsequent_indent
|
||||
else:
|
||||
indent = self.initial_indent
|
||||
if term_len(indent) + term_len(self.placeholder.lstrip()) > self.width:
|
||||
raise ValueError("placeholder too large for max width")
|
||||
|
||||
chunks.reverse()
|
||||
|
||||
while chunks:
|
||||
cur_line: list[str] = []
|
||||
cur_len = 0
|
||||
|
||||
if lines:
|
||||
indent = self.subsequent_indent
|
||||
else:
|
||||
indent = self.initial_indent
|
||||
|
||||
width = self.width - term_len(indent)
|
||||
|
||||
if self.drop_whitespace and chunks[-1].strip() == "" and lines:
|
||||
del chunks[-1]
|
||||
|
||||
while chunks:
|
||||
n = term_len(chunks[-1])
|
||||
|
||||
if cur_len + n <= width:
|
||||
cur_line.append(chunks.pop())
|
||||
cur_len += n
|
||||
|
||||
else:
|
||||
break
|
||||
|
||||
if chunks and term_len(chunks[-1]) > width:
|
||||
self._handle_long_word(chunks, cur_line, cur_len, width)
|
||||
cur_len = sum(map(term_len, cur_line))
|
||||
|
||||
if self.drop_whitespace and cur_line and cur_line[-1].strip() == "":
|
||||
cur_len -= term_len(cur_line[-1])
|
||||
del cur_line[-1]
|
||||
|
||||
if cur_line:
|
||||
if (
|
||||
self.max_lines is None
|
||||
or len(lines) + 1 < self.max_lines
|
||||
or (
|
||||
not chunks
|
||||
or self.drop_whitespace
|
||||
and len(chunks) == 1
|
||||
and not chunks[0].strip()
|
||||
)
|
||||
and cur_len <= width
|
||||
):
|
||||
lines.append(indent + "".join(cur_line))
|
||||
else:
|
||||
while cur_line:
|
||||
if (
|
||||
cur_line[-1].strip()
|
||||
and cur_len + term_len(self.placeholder) <= width
|
||||
):
|
||||
cur_line.append(self.placeholder)
|
||||
lines.append(indent + "".join(cur_line))
|
||||
break
|
||||
cur_len -= term_len(cur_line[-1])
|
||||
del cur_line[-1]
|
||||
else:
|
||||
if lines:
|
||||
prev_line = lines[-1].rstrip()
|
||||
if (
|
||||
term_len(prev_line) + term_len(self.placeholder)
|
||||
<= self.width
|
||||
):
|
||||
lines[-1] = prev_line + self.placeholder
|
||||
break
|
||||
lines.append(indent + self.placeholder.lstrip())
|
||||
break
|
||||
|
||||
return lines
|
||||
|
||||
@contextmanager
|
||||
def extra_indent(self, indent: str) -> cabc.Iterator[None]:
|
||||
old_initial_indent = self.initial_indent
|
||||
old_subsequent_indent = self.subsequent_indent
|
||||
self.initial_indent += indent
|
||||
self.subsequent_indent += indent
|
||||
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
self.initial_indent = old_initial_indent
|
||||
self.subsequent_indent = old_subsequent_indent
|
||||
|
||||
def indent_only(self, text: str) -> str:
|
||||
rv = []
|
||||
|
||||
for idx, line in enumerate(text.splitlines()):
|
||||
indent = self.initial_indent
|
||||
|
||||
if idx > 0:
|
||||
indent = self.subsequent_indent
|
||||
|
||||
rv.append(f"{indent}{line}")
|
||||
|
||||
return "\n".join(rv)
|
||||
36
venv/Lib/site-packages/click/_utils.py
Normal file
36
venv/Lib/site-packages/click/_utils.py
Normal file
@@ -0,0 +1,36 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import enum
|
||||
import typing as t
|
||||
|
||||
|
||||
class Sentinel(enum.Enum):
|
||||
"""Enum used to define sentinel values.
|
||||
|
||||
.. seealso::
|
||||
|
||||
`PEP 661 - Sentinel Values <https://peps.python.org/pep-0661/>`_.
|
||||
"""
|
||||
|
||||
UNSET = object()
|
||||
FLAG_NEEDS_VALUE = object()
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}.{self.name}"
|
||||
|
||||
|
||||
UNSET: t.Literal[Sentinel.UNSET] = Sentinel.UNSET
|
||||
"""Sentinel used to indicate that a value is not set."""
|
||||
|
||||
FLAG_NEEDS_VALUE: t.Literal[Sentinel.FLAG_NEEDS_VALUE] = Sentinel.FLAG_NEEDS_VALUE
|
||||
"""Sentinel used to indicate an option was passed as a flag without a
|
||||
value but is not a flag option.
|
||||
|
||||
``Option.consume_value`` uses this to prompt or use the ``flag_value``.
|
||||
"""
|
||||
|
||||
T_UNSET: t.TypeAlias = t.Literal[Sentinel.UNSET]
|
||||
"""Type hint for the :data:`UNSET` sentinel value."""
|
||||
|
||||
T_FLAG_NEEDS_VALUE: t.TypeAlias = t.Literal[Sentinel.FLAG_NEEDS_VALUE]
|
||||
"""Type hint for the :data:`FLAG_NEEDS_VALUE` sentinel value."""
|
||||
Reference in New Issue
Block a user