back to Reference (Gold) summary
Reference (Gold): click
Pytest Summary for test tests
| status | count |
|---|---|
| passed | 589 |
| xfailed | 1 |
| skipped | 21 |
| total | 611 |
| collected | 611 |
Failed pytests:
test_chain.py::test_multicommand_chaining
test_chain.py::test_multicommand_chaining
runner =@pytest.mark.xfail def test_multicommand_chaining(runner): @click.group(chain=True) def cli(): debug() @cli.group() > def l1a(): tests/test_chain.py:228: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ src/click/core.py:1942: in decorator self.add_command(cmd) src/click/core.py:1842: in add_command _check_multicommand(self, name, cmd, register=True) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ base_command = , cmd_name = 'l1a', cmd = , register = True def _check_multicommand( base_command: "MultiCommand", cmd_name: str, cmd: "Command", register: bool = False ) -> None: if not base_command.chain or not isinstance(cmd, MultiCommand): return if register: hint = ( "It is not possible to add multi commands as children to" " another multi command that is in chain mode." ) else: hint = ( "Found a multi command as subcommand to a multi command" " that is in chain mode. This is not supported." ) > raise RuntimeError( f"{hint}. Command {base_command.name!r} is set to chain and" f" {cmd_name!r} was added as a subcommand but it in itself is a" f" multi command. ({cmd_name!r} is a {type(cmd).__name__}" f" within a chained {type(base_command).__name__} named" f" {base_command.name!r})." ) E RuntimeError: It is not possible to add multi commands as children to another multi command that is in chain mode.. Command 'cli' is set to chain and 'l1a' was added as a subcommand but it in itself is a multi command. ('l1a' is a Group within a chained Group named 'cli'). src/click/core.py:82: RuntimeError
Patch diff
diff --git a/src/click/_compat.py b/src/click/_compat.py
index 7f5b9af..23f8866 100644
--- a/src/click/_compat.py
+++ b/src/click/_compat.py
@@ -5,37 +5,75 @@ import re
import sys
import typing as t
from weakref import WeakKeyDictionary
-CYGWIN = sys.platform.startswith('cygwin')
-WIN = sys.platform.startswith('win')
+
+CYGWIN = sys.platform.startswith("cygwin")
+WIN = sys.platform.startswith("win")
auto_wrap_for_ansi: t.Optional[t.Callable[[t.TextIO], t.TextIO]] = None
-_ansi_re = re.compile('\\033\\[[;?0-9]*[a-zA-Z]')
+_ansi_re = re.compile(r"\033\[[;?0-9]*[a-zA-Z]")
+
+def _make_text_stream(
+ stream: t.BinaryIO,
+ encoding: t.Optional[str],
+ errors: t.Optional[str],
+ 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:
+
+def is_ascii_encoding(encoding: str) -> bool:
"""Checks if a given encoding is ascii."""
- pass
+ try:
+ return codecs.lookup(encoding).name == "ascii"
+ except LookupError:
+ return False
-def get_best_encoding(stream: t.IO[t.Any]) ->str:
+def get_best_encoding(stream: t.IO[t.Any]) -> str:
"""Returns the default stream encoding if not found."""
- pass
+ 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: t.Optional[str],
- errors: t.Optional[str], 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))
+ def __init__(
+ self,
+ stream: t.BinaryIO,
+ encoding: t.Optional[str],
+ errors: t.Optional[str],
+ 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:
+ 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
@@ -47,86 +85,539 @@ class _FixupStream:
of jupyter notebook).
"""
- def __init__(self, stream: t.BinaryIO, force_readable: bool=False,
- force_writable: bool=False):
+ 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:
+ 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("") # type: ignore
+ 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.Optional[t.BinaryIO]:
+ # 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.Optional[t.BinaryIO]:
+ # 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)
-def _stream_is_misconfigured(stream: t.TextIO) ->bool:
+ 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."""
- pass
+ # 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: t.Optional[str]
- ) ->bool:
+def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> 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.
"""
- pass
+ 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: t.Optional[str],
- errors: t.Optional[str]) ->bool:
+def _is_compatible_text_stream(
+ stream: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
+) -> bool:
"""Check if a stream's encoding and errors attributes are
compatible with the desired values.
"""
- pass
+ 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: t.Optional[str],
+ errors: t.Optional[str],
+ is_binary: t.Callable[[t.IO[t.Any], bool], bool],
+ find_binary: t.Callable[[t.IO[t.Any]], t.Optional[t.BinaryIO]],
+ 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: t.Optional[str],
+ errors: t.Optional[str],
+ 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: t.Optional[str],
+ errors: t.Optional[str],
+ 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 _wrap_io_open(file: t.Union[str, 'os.PathLike[str]', int], mode: str,
- encoding: t.Optional[str], errors: t.Optional[str]) ->t.IO[t.Any]:
+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: t.Optional[str] = None, errors: t.Optional[str] = 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: t.Optional[str] = None, errors: t.Optional[str] = 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: t.Optional[str] = None, errors: t.Optional[str] = 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: t.Union[str, "os.PathLike[str]", int],
+ mode: str,
+ encoding: t.Optional[str],
+ errors: t.Optional[str],
+) -> t.IO[t.Any]:
"""Handles not passing ``encoding`` and ``errors`` in binary mode."""
- pass
+ if "b" in mode:
+ return open(file, mode)
+ return open(file, mode, encoding=encoding, errors=errors)
-class _AtomicFile:
- def __init__(self, f: t.IO[t.Any], tmp_filename: str, real_filename: str
- ) ->None:
+def open_stream(
+ filename: "t.Union[str, os.PathLike[str]]",
+ mode: str = "r",
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+ atomic: bool = False,
+) -> t.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: t.Optional[int] = 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
- def __getattr__(self, name: str) ->t.Any:
+ @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':
+ def __enter__(self) -> "_AtomicFile":
return self
- def __exit__(self, exc_type: t.Optional[t.Type[BaseException]], *_: t.Any
- ) ->None:
+ def __exit__(self, exc_type: t.Optional[t.Type[BaseException]], *_: t.Any) -> None:
self.close(delete=exc_type is not None)
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
return repr(self._f)
-if sys.platform.startswith('win') and WIN:
+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.Optional[t.IO[t.Any]] = None, color: t.Optional[bool] = None
+) -> bool:
+ if color is None:
+ if stream is None:
+ stream = sys.stdin
+ 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
- _ansi_stream_wrappers: t.MutableMapping[t.TextIO, t.TextIO
- ] = WeakKeyDictionary()
- def auto_wrap_for_ansi(stream: t.TextIO, color: t.Optional[bool]=None
- ) ->t.TextIO:
+ def _get_argv_encoding() -> str:
+ import locale
+
+ return locale.getpreferredencoding()
+
+ _ansi_stream_wrappers: t.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()
+
+ def auto_wrap_for_ansi( # noqa: F811
+ stream: t.TextIO, color: t.Optional[bool] = None
+ ) -> t.TextIO:
"""Support ANSI color and style codes on Windows by wrapping a
stream with colorama.
"""
- pass
-_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: t.Mapping[str, t.Callable[[], t.BinaryIO]] = {'stdin':
- get_binary_stdin, 'stdout': get_binary_stdout, 'stderr': get_binary_stderr}
-text_streams: t.Mapping[str, t.Callable[[t.Optional[str], t.Optional[str]],
- t.TextIO]] = {'stdin': get_text_stdin, 'stdout': get_text_stdout,
- 'stderr': get_text_stderr}
+ 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):
+ try:
+ return _write(s)
+ except BaseException:
+ ansi_wrapper.reset_all()
+ raise
+
+ rv.write = _safe_write
+
+ 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: t.Optional[str], errors: t.Optional[str]
+ ) -> t.Optional[t.TextIO]:
+ 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.Optional[t.TextIO]],
+ wrapper_func: t.Callable[[], t.TextIO],
+) -> t.Callable[[], t.Optional[t.TextIO]]:
+ cache: t.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()
+
+ def func() -> t.Optional[t.TextIO]:
+ 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: t.Mapping[str, t.Callable[[], t.BinaryIO]] = {
+ "stdin": get_binary_stdin,
+ "stdout": get_binary_stdout,
+ "stderr": get_binary_stderr,
+}
+
+text_streams: t.Mapping[
+ str, t.Callable[[t.Optional[str], t.Optional[str]], t.TextIO]
+] = {
+ "stdin": get_text_stdin,
+ "stdout": get_text_stdout,
+ "stderr": get_text_stderr,
+}
diff --git a/src/click/_termui_impl.py b/src/click/_termui_impl.py
index 4ed4097..f744657 100644
--- a/src/click/_termui_impl.py
+++ b/src/click/_termui_impl.py
@@ -12,6 +12,7 @@ import typing as t
from gettext import gettext as _
from io import StringIO
from types import TracebackType
+
from ._compat import _default_text_stdout
from ._compat import CYGWIN
from ._compat import get_best_encoding
@@ -22,25 +23,36 @@ from ._compat import term_len
from ._compat import WIN
from .exceptions import ClickException
from .utils import echo
-V = t.TypeVar('V')
-if os.name == 'nt':
- BEFORE_BAR = '\r'
- AFTER_BAR = '\n'
+
+V = t.TypeVar("V")
+
+if os.name == "nt":
+ BEFORE_BAR = "\r"
+ AFTER_BAR = "\n"
else:
- BEFORE_BAR = '\r\x1b[?25l'
- AFTER_BAR = '\x1b[?25h\n'
+ BEFORE_BAR = "\r\033[?25l"
+ AFTER_BAR = "\033[?25h\n"
class ProgressBar(t.Generic[V]):
-
- def __init__(self, iterable: t.Optional[t.Iterable[V]], length: t.
- Optional[int]=None, fill_char: str='#', empty_char: str=' ',
- bar_template: str='%(bar)s', info_sep: str=' ', show_eta: bool=
- True, show_percent: t.Optional[bool]=None, show_pos: bool=False,
- item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[
- str]]]=None, label: t.Optional[str]=None, file: t.Optional[t.TextIO
- ]=None, color: t.Optional[bool]=None, update_min_steps: int=1,
- width: int=30) ->None:
+ def __init__(
+ self,
+ iterable: t.Optional[t.Iterable[V]],
+ length: t.Optional[int] = None,
+ fill_char: str = "#",
+ empty_char: str = " ",
+ bar_template: str = "%(bar)s",
+ info_sep: str = " ",
+ show_eta: bool = True,
+ show_percent: t.Optional[bool] = None,
+ show_pos: bool = False,
+ item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None,
+ label: t.Optional[str] = None,
+ file: t.Optional[t.TextIO] = None,
+ color: t.Optional[bool] = 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
@@ -49,25 +61,33 @@ class ProgressBar(t.Generic[V]):
self.show_percent = show_percent
self.show_pos = show_pos
self.item_show_func = item_show_func
- self.label: str = label or ''
+ 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')
+ raise TypeError("iterable or length is required")
iterable = t.cast(t.Iterable[V], range(length))
self.iter: t.Iterable[V] = iter(iterable)
self.length = length
@@ -84,27 +104,195 @@ class ProgressBar(t.Generic[V]):
self.is_hidden: bool = not isatty(self.file)
self._last_line: t.Optional[str] = None
- def __enter__(self) ->'ProgressBar[V]':
+ def __enter__(self) -> "ProgressBar[V]":
self.entered = True
self.render_progress()
return self
- def __exit__(self, exc_type: t.Optional[t.Type[BaseException]],
- exc_value: t.Optional[BaseException], tb: t.Optional[TracebackType]
- ) ->None:
+ def __exit__(
+ self,
+ exc_type: t.Optional[t.Type[BaseException]],
+ exc_value: t.Optional[BaseException],
+ tb: t.Optional[TracebackType],
+ ) -> None:
self.render_finish()
- def __iter__(self) ->t.Iterator[V]:
+ def __iter__(self) -> t.Iterator[V]:
if not self.entered:
- raise RuntimeError('You need to use progress bars in a with block.'
- )
+ raise RuntimeError("You need to use progress bars in a with block.")
self.render_progress()
return self.generator()
- def __next__(self) ->V:
+ 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 update(self, n_steps: int, current_item: t.Optional[V]=None) ->None:
+ def render_finish(self) -> None:
+ if self.is_hidden:
+ 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 f"{t}d {hours:02}:{minutes:02}:{seconds:02}"
+ 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:
+ import shutil
+
+ if self.is_hidden:
+ # Only output the label as it changes if the output is not a
+ # TTY. Use file=stderr if you expect to be piping stdout.
+ 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:
+ 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:
+ buf.append(BEFORE_BAR)
+ buf.append(" " * self.max_width) # type: ignore
+ 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: t.Optional[V] = None) -> None:
"""Update the progress bar by advancing a specified number of
steps, and optionally set the ``current_item`` for this new
position.
@@ -120,54 +308,432 @@ class ProgressBar(t.Generic[V]):
Only render when the number of steps meets the
``update_min_steps`` threshold.
"""
- pass
+ if current_item is not None:
+ self.current_item = current_item
+
+ self._completed_intervals += n_steps
- def generator(self) ->t.Iterator[V]:
+ 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) -> t.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.
"""
- pass
+ # 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 self.is_hidden:
+ 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()
-def pager(generator: t.Iterable[str], color: t.Optional[bool]=None) ->None:
+
+def pager(generator: t.Iterable[str], color: t.Optional[bool] = None) -> None:
"""Decide what method to use for paging through text."""
- pass
+ 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, generator, color)
+ pager_cmd = (os.environ.get("PAGER", None) or "").strip()
+ if pager_cmd:
+ if WIN:
+ return _tempfilepager(generator, pager_cmd, color)
+ return _pipepager(generator, pager_cmd, color)
+ if os.environ.get("TERM") in ("dumb", "emacs"):
+ return _nullpager(stdout, generator, color)
+ if WIN or sys.platform.startswith("os2"):
+ return _tempfilepager(generator, "more <", color)
+ if hasattr(os, "system") and os.system("(less) 2>/dev/null") == 0:
+ return _pipepager(generator, "less", color)
+ import tempfile
-def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]
- ) ->None:
+ fd, filename = tempfile.mkstemp()
+ os.close(fd)
+ try:
+ if hasattr(os, "system") and os.system(f'more "{filename}"') == 0:
+ return _pipepager(generator, "more", color)
+ return _nullpager(stdout, generator, color)
+ finally:
+ os.unlink(filename)
+
+
+def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]) -> None:
"""Page through text by feeding it to another program. Invoking a
pager through this might support colors.
"""
- pass
+ import subprocess
+
+ env = dict(os.environ)
+ # If we're piping to less we might support colors under the
+ # condition that
+ cmd_detail = cmd.rsplit("/", 1)[-1].split()
+ if color is None and cmd_detail[0] == "less":
+ less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_detail[1:])}"
+ if not less_flags:
+ env["LESS"] = "-R"
+ color = True
+ elif "r" in less_flags or "R" in less_flags:
+ color = True
-def _tempfilepager(generator: t.Iterable[str], cmd: str, color: t.Optional[
- bool]) ->None:
+ c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env)
+ stdin = t.cast(t.BinaryIO, c.stdin)
+ encoding = get_best_encoding(stdin)
+ try:
+ for text in generator:
+ if not color:
+ text = strip_ansi(text)
+
+ stdin.write(text.encode(encoding, "replace"))
+ except (OSError, KeyboardInterrupt):
+ pass
+ else:
+ stdin.close()
+
+ # 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
+
+
+def _tempfilepager(
+ generator: t.Iterable[str], cmd: str, color: t.Optional[bool]
+) -> None:
"""Page through text by invoking a program on a temporary file."""
- pass
+ import tempfile
+
+ fd, filename = tempfile.mkstemp()
+ # TODO: This never terminates if the passed generator never terminates.
+ text = "".join(generator)
+ if not color:
+ text = strip_ansi(text)
+ encoding = get_best_encoding(sys.stdout)
+ with open_stream(filename, "wb")[0] as f:
+ f.write(text.encode(encoding))
+ try:
+ os.system(f'{cmd} "{filename}"')
+ finally:
+ os.close(fd)
+ os.unlink(filename)
-def _nullpager(stream: t.TextIO, generator: t.Iterable[str], color: t.
- Optional[bool]) ->None:
+def _nullpager(
+ stream: t.TextIO, generator: t.Iterable[str], color: t.Optional[bool]
+) -> None:
"""Simply print unformatted text. This is the ultimate fallback."""
- pass
+ for text in generator:
+ if not color:
+ text = strip_ansi(text)
+ stream.write(text)
class Editor:
-
- def __init__(self, editor: t.Optional[str]=None, env: t.Optional[t.
- Mapping[str, str]]=None, require_save: bool=True, extension: str='.txt'
- ) ->None:
+ def __init__(
+ self,
+ editor: t.Optional[str] = None,
+ env: t.Optional[t.Mapping[str, str]] = 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"
+ for editor in "sensible-editor", "vim", "nano":
+ if os.system(f"which {editor} >/dev/null 2>&1") == 0:
+ return editor
+ return "vi"
+
+ def edit_file(self, filename: str) -> None:
+ import subprocess
+
+ editor = self.get_editor()
+ environ: t.Optional[t.Dict[str, str]] = None
+
+ if self.env:
+ environ = os.environ.copy()
+ environ.update(self.env)
+
+ try:
+ c = subprocess.Popen(f'{editor} "{filename}"', env=environ, shell=True)
+ 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
+
+ def edit(self, text: t.Optional[t.AnyStr]) -> t.Optional[t.AnyStr]:
+ import tempfile
+
+ if not text:
+ data = 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_file(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") # type: ignore
+ 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.replace('"', ""))
+ args = f'explorer /select,"{url}"'
+ else:
+ url = url.replace('"', "")
+ wait_str = "/WAIT" if wait else ""
+ args = f'start {wait_str} "" "{url}"'
+ return os.system(args)
+ elif CYGWIN:
+ if locate:
+ url = os.path.dirname(_unquote_file(url).replace('"', ""))
+ args = f'cygstart "{url}"'
+ else:
+ url = url.replace('"', "")
+ wait_str = "-w" if wait else ""
+ args = f'cygstart {wait_str} "{url}"'
+ return os.system(args)
+
+ 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) -> t.Optional[BaseException]:
+ 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()
+
+ return None
+
if WIN:
import msvcrt
+
+ @contextlib.contextmanager
+ def raw_terminal() -> t.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`.
+ func: t.Callable[[], str]
+
+ if echo:
+ func = msvcrt.getwche # type: ignore
+ else:
+ func = msvcrt.getwch # type: ignore
+
+ 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 tty
import termios
+
+ @contextlib.contextmanager
+ def raw_terminal() -> t.Iterator[int]:
+ f: t.Optional[t.TextIO]
+ 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
diff --git a/src/click/_textwrap.py b/src/click/_textwrap.py
index 681ea25..b47dcbd 100644
--- a/src/click/_textwrap.py
+++ b/src/click/_textwrap.py
@@ -4,4 +4,46 @@ from contextlib import contextmanager
class TextWrapper(textwrap.TextWrapper):
- pass
+ def _handle_long_word(
+ self,
+ reversed_chunks: t.List[str],
+ cur_line: t.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 = last[:space_left]
+ res = last[space_left:]
+ cur_line.append(cut)
+ reversed_chunks[-1] = res
+ elif not cur_line:
+ cur_line.append(reversed_chunks.pop())
+
+ @contextmanager
+ def extra_indent(self, indent: str) -> t.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)
diff --git a/src/click/_winconsole.py b/src/click/_winconsole.py
index 514d97e..6b20df3 100644
--- a/src/click/_winconsole.py
+++ b/src/click/_winconsole.py
@@ -1,3 +1,11 @@
+# This module is based on the excellent work by Adam Bartoš who
+# provided a lot of what went into the implementation here in
+# the discussion to issue1602 in the Python bug tracker.
+#
+# There are some general differences in regards to how this works
+# compared to the original patches as we do not need to patch
+# the entire interpreter but just work in our little world of
+# echo and prompt.
import io
import sys
import time
@@ -16,78 +24,256 @@ from ctypes.wintypes import DWORD
from ctypes.wintypes import HANDLE
from ctypes.wintypes import LPCWSTR
from ctypes.wintypes import LPWSTR
+
from ._compat import _NonClosingTextIOWrapper
-assert sys.platform == 'win32'
-import msvcrt
-from ctypes import windll
-from ctypes import WINFUNCTYPE
+
+assert sys.platform == "win32"
+import msvcrt # noqa: E402
+from ctypes import windll # noqa: E402
+from ctypes import WINFUNCTYPE # noqa: E402
+
c_ssize_p = POINTER(c_ssize_t)
+
kernel32 = windll.kernel32
GetStdHandle = kernel32.GetStdHandle
ReadConsoleW = kernel32.ReadConsoleW
WriteConsoleW = kernel32.WriteConsoleW
GetConsoleMode = kernel32.GetConsoleMode
GetLastError = kernel32.GetLastError
-GetCommandLineW = WINFUNCTYPE(LPWSTR)(('GetCommandLineW', windll.kernel32))
-CommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))((
- 'CommandLineToArgvW', windll.shell32))
-LocalFree = WINFUNCTYPE(c_void_p, c_void_p)(('LocalFree', windll.kernel32))
+GetCommandLineW = WINFUNCTYPE(LPWSTR)(("GetCommandLineW", windll.kernel32))
+CommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))(
+ ("CommandLineToArgvW", windll.shell32)
+)
+LocalFree = WINFUNCTYPE(c_void_p, c_void_p)(("LocalFree", windll.kernel32))
+
STDIN_HANDLE = GetStdHandle(-10)
STDOUT_HANDLE = GetStdHandle(-11)
STDERR_HANDLE = GetStdHandle(-12)
+
PyBUF_SIMPLE = 0
PyBUF_WRITABLE = 1
+
ERROR_SUCCESS = 0
ERROR_NOT_ENOUGH_MEMORY = 8
ERROR_OPERATION_ABORTED = 995
+
STDIN_FILENO = 0
STDOUT_FILENO = 1
STDERR_FILENO = 2
-EOF = b'\x1a'
+
+EOF = b"\x1a"
MAX_BYTES_WRITTEN = 32767
+
try:
from ctypes import pythonapi
except ImportError:
+ # On PyPy we cannot get buffers so our ability to operate here is
+ # severely limited.
get_buffer = None
else:
-
class Py_buffer(Structure):
- _fields_ = [('buf', c_void_p), ('obj', py_object), ('len',
- c_ssize_t), ('itemsize', c_ssize_t), ('readonly', c_int), (
- 'ndim', c_int), ('format', c_char_p), ('shape', c_ssize_p), (
- 'strides', c_ssize_p), ('suboffsets', c_ssize_p), ('internal',
- c_void_p)]
+ _fields_ = [
+ ("buf", c_void_p),
+ ("obj", py_object),
+ ("len", c_ssize_t),
+ ("itemsize", c_ssize_t),
+ ("readonly", c_int),
+ ("ndim", c_int),
+ ("format", c_char_p),
+ ("shape", c_ssize_p),
+ ("strides", c_ssize_p),
+ ("suboffsets", c_ssize_p),
+ ("internal", c_void_p),
+ ]
+
PyObject_GetBuffer = pythonapi.PyObject_GetBuffer
PyBuffer_Release = pythonapi.PyBuffer_Release
+ def get_buffer(obj, writable=False):
+ buf = Py_buffer()
+ flags = PyBUF_WRITABLE if writable else PyBUF_SIMPLE
+ PyObject_GetBuffer(py_object(obj), byref(buf), flags)
+
+ try:
+ buffer_type = c_char * buf.len
+ return buffer_type.from_address(buf.buf)
+ finally:
+ PyBuffer_Release(byref(buf))
-class _WindowsConsoleRawIOBase(io.RawIOBase):
+class _WindowsConsoleRawIOBase(io.RawIOBase):
def __init__(self, handle):
self.handle = handle
+ def isatty(self):
+ super().isatty()
+ return True
+
class _WindowsConsoleReader(_WindowsConsoleRawIOBase):
- pass
+ def readable(self):
+ return True
+
+ def readinto(self, b):
+ bytes_to_be_read = len(b)
+ if not bytes_to_be_read:
+ return 0
+ elif bytes_to_be_read % 2:
+ raise ValueError(
+ "cannot read odd number of bytes from UTF-16-LE encoded console"
+ )
+
+ buffer = get_buffer(b, writable=True)
+ code_units_to_be_read = bytes_to_be_read // 2
+ code_units_read = c_ulong()
+
+ rv = ReadConsoleW(
+ HANDLE(self.handle),
+ buffer,
+ code_units_to_be_read,
+ byref(code_units_read),
+ None,
+ )
+ if GetLastError() == ERROR_OPERATION_ABORTED:
+ # wait for KeyboardInterrupt
+ time.sleep(0.1)
+ if not rv:
+ raise OSError(f"Windows error: {GetLastError()}")
+
+ if buffer[0] == EOF:
+ return 0
+ return 2 * code_units_read.value
class _WindowsConsoleWriter(_WindowsConsoleRawIOBase):
- pass
+ def writable(self):
+ return True
+ @staticmethod
+ def _get_error_message(errno):
+ if errno == ERROR_SUCCESS:
+ return "ERROR_SUCCESS"
+ elif errno == ERROR_NOT_ENOUGH_MEMORY:
+ return "ERROR_NOT_ENOUGH_MEMORY"
+ return f"Windows error {errno}"
-class ConsoleStream:
+ def write(self, b):
+ bytes_to_be_written = len(b)
+ buf = get_buffer(b)
+ code_units_to_be_written = min(bytes_to_be_written, MAX_BYTES_WRITTEN) // 2
+ code_units_written = c_ulong()
+
+ WriteConsoleW(
+ HANDLE(self.handle),
+ buf,
+ code_units_to_be_written,
+ byref(code_units_written),
+ None,
+ )
+ bytes_written = 2 * code_units_written.value
- def __init__(self, text_stream: t.TextIO, byte_stream: t.BinaryIO) ->None:
+ if bytes_written == 0 and bytes_to_be_written > 0:
+ raise OSError(self._get_error_message(GetLastError()))
+ return bytes_written
+
+
+class ConsoleStream:
+ def __init__(self, text_stream: t.TextIO, byte_stream: t.BinaryIO) -> None:
self._text_stream = text_stream
self.buffer = byte_stream
- def __getattr__(self, name: str) ->t.Any:
+ @property
+ def name(self) -> str:
+ return self.buffer.name
+
+ def write(self, x: t.AnyStr) -> int:
+ if isinstance(x, str):
+ return self._text_stream.write(x)
+ try:
+ self.flush()
+ except Exception:
+ pass
+ return self.buffer.write(x)
+
+ def writelines(self, lines: t.Iterable[t.AnyStr]) -> None:
+ for line in lines:
+ self.write(line)
+
+ def __getattr__(self, name: str) -> t.Any:
return getattr(self._text_stream, name)
+ def isatty(self) -> bool:
+ return self.buffer.isatty()
+
def __repr__(self):
- return f'<ConsoleStream name={self.name!r} encoding={self.encoding!r}>'
+ return f"<ConsoleStream name={self.name!r} encoding={self.encoding!r}>"
+
+
+def _get_text_stdin(buffer_stream: t.BinaryIO) -> t.TextIO:
+ text_stream = _NonClosingTextIOWrapper(
+ io.BufferedReader(_WindowsConsoleReader(STDIN_HANDLE)),
+ "utf-16-le",
+ "strict",
+ line_buffering=True,
+ )
+ return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))
+
+
+def _get_text_stdout(buffer_stream: t.BinaryIO) -> t.TextIO:
+ text_stream = _NonClosingTextIOWrapper(
+ io.BufferedWriter(_WindowsConsoleWriter(STDOUT_HANDLE)),
+ "utf-16-le",
+ "strict",
+ line_buffering=True,
+ )
+ return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))
+
+
+def _get_text_stderr(buffer_stream: t.BinaryIO) -> t.TextIO:
+ text_stream = _NonClosingTextIOWrapper(
+ io.BufferedWriter(_WindowsConsoleWriter(STDERR_HANDLE)),
+ "utf-16-le",
+ "strict",
+ line_buffering=True,
+ )
+ return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))
+
+
+_stream_factories: t.Mapping[int, t.Callable[[t.BinaryIO], t.TextIO]] = {
+ 0: _get_text_stdin,
+ 1: _get_text_stdout,
+ 2: _get_text_stderr,
+}
+
+
+def _is_console(f: t.TextIO) -> bool:
+ if not hasattr(f, "fileno"):
+ return False
+
+ try:
+ fileno = f.fileno()
+ except (OSError, io.UnsupportedOperation):
+ return False
+
+ handle = msvcrt.get_osfhandle(fileno)
+ return bool(GetConsoleMode(handle, byref(DWORD())))
+
+
+def _get_windows_console_stream(
+ f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
+) -> t.Optional[t.TextIO]:
+ if (
+ get_buffer is not None
+ and encoding in {"utf-16-le", None}
+ and errors in {"strict", None}
+ and _is_console(f)
+ ):
+ func = _stream_factories.get(f.fileno())
+ if func is not None:
+ b = getattr(f, "buffer", None)
+ if b is None:
+ return None
-_stream_factories: t.Mapping[int, t.Callable[[t.BinaryIO], t.TextIO]] = {(0
- ): _get_text_stdin, (1): _get_text_stdout, (2): _get_text_stderr}
+ return func(b)
diff --git a/src/click/core.py b/src/click/core.py
index 354452a..cc65e89 100644
--- a/src/click/core.py
+++ b/src/click/core.py
@@ -12,6 +12,7 @@ from gettext import gettext as _
from gettext import ngettext
from itertools import repeat
from types import TracebackType
+
from . import types
from .exceptions import Abort
from .exceptions import BadParameter
@@ -35,38 +36,99 @@ from .utils import echo
from .utils import make_default_short_help
from .utils import make_str
from .utils import PacifyFlushWrapper
+
if t.TYPE_CHECKING:
import typing_extensions as te
from .shell_completion import CompletionItem
-F = t.TypeVar('F', bound=t.Callable[..., t.Any])
-V = t.TypeVar('V')
+
+F = t.TypeVar("F", bound=t.Callable[..., t.Any])
+V = t.TypeVar("V")
-def _complete_visible_commands(ctx: 'Context', incomplete: str) ->t.Iterator[t
- .Tuple[str, 'Command']]:
+def _complete_visible_commands(
+ ctx: "Context", incomplete: str
+) -> t.Iterator[t.Tuple[str, "Command"]]:
"""List all the subcommands of a group that start with the
incomplete value and aren't hidden.
:param ctx: Invocation context for the group.
:param incomplete: Value being completed. May be empty.
"""
- pass
+ multi = t.cast(MultiCommand, ctx.command)
+
+ for name in multi.list_commands(ctx):
+ if name.startswith(incomplete):
+ command = multi.get_command(ctx, name)
+
+ if command is not None and not command.hidden:
+ yield name, command
+
+
+def _check_multicommand(
+ base_command: "MultiCommand", cmd_name: str, cmd: "Command", register: bool = False
+) -> None:
+ if not base_command.chain or not isinstance(cmd, MultiCommand):
+ return
+ if register:
+ hint = (
+ "It is not possible to add multi commands as children to"
+ " another multi command that is in chain mode."
+ )
+ else:
+ hint = (
+ "Found a multi command as subcommand to a multi command"
+ " that is in chain mode. This is not supported."
+ )
+ raise RuntimeError(
+ f"{hint}. Command {base_command.name!r} is set to chain and"
+ f" {cmd_name!r} was added as a subcommand but it in itself is a"
+ f" multi command. ({cmd_name!r} is a {type(cmd).__name__}"
+ f" within a chained {type(base_command).__name__} named"
+ f" {base_command.name!r})."
+ )
+
+
+def batch(iterable: t.Iterable[V], batch_size: int) -> t.List[t.Tuple[V, ...]]:
+ return list(zip(*repeat(iter(iterable), batch_size)))
@contextmanager
-def augment_usage_errors(ctx: 'Context', param: t.Optional['Parameter']=None
- ) ->t.Iterator[None]:
+def augment_usage_errors(
+ ctx: "Context", param: t.Optional["Parameter"] = None
+) -> t.Iterator[None]:
"""Context manager that attaches extra information to exceptions."""
- pass
-
-
-def iter_params_for_processing(invocation_order: t.Sequence['Parameter'],
- declaration_order: t.Sequence['Parameter']) ->t.List['Parameter']:
+ try:
+ yield
+ except BadParameter as e:
+ if e.ctx is None:
+ e.ctx = ctx
+ if param is not None and e.param is None:
+ e.param = param
+ raise
+ except UsageError as e:
+ if e.ctx is None:
+ e.ctx = ctx
+ raise
+
+
+def iter_params_for_processing(
+ invocation_order: t.Sequence["Parameter"],
+ declaration_order: t.Sequence["Parameter"],
+) -> t.List["Parameter"]:
"""Given a sequence of parameters in the order as should be considered
for processing and an iterable of parameters that exist, this returns
a list in the correct order as they should be processed.
"""
- pass
+
+ def sort_key(item: "Parameter") -> t.Tuple[bool, float]:
+ try:
+ idx: float = invocation_order.index(item)
+ except ValueError:
+ idx = float("inf")
+
+ return not item.is_eager, idx
+
+ return sorted(declaration_order, key=sort_key)
class ParameterSource(enum.Enum):
@@ -82,6 +144,7 @@ class ParameterSource(enum.Enum):
.. versionchanged:: 8.0
Added the ``PROMPT`` value.
"""
+
COMMANDLINE = enum.auto()
"""The value was provided by the command line args."""
ENVIRONMENT = enum.auto()
@@ -188,85 +251,185 @@ class Context:
Added the ``resilient_parsing``, ``help_option_names``, and
``token_normalize_func`` parameters.
"""
- formatter_class: t.Type['HelpFormatter'] = HelpFormatter
-
- def __init__(self, command: 'Command', parent: t.Optional['Context']=
- None, info_name: t.Optional[str]=None, obj: t.Optional[t.Any]=None,
- auto_envvar_prefix: t.Optional[str]=None, default_map: t.Optional[t
- .MutableMapping[str, t.Any]]=None, terminal_width: t.Optional[int]=
- None, max_content_width: t.Optional[int]=None, resilient_parsing:
- bool=False, allow_extra_args: t.Optional[bool]=None,
- allow_interspersed_args: t.Optional[bool]=None,
- ignore_unknown_options: t.Optional[bool]=None, help_option_names: t
- .Optional[t.List[str]]=None, token_normalize_func: t.Optional[t.
- Callable[[str], str]]=None, color: t.Optional[bool]=None,
- show_default: t.Optional[bool]=None) ->None:
+
+ #: The formatter class to create with :meth:`make_formatter`.
+ #:
+ #: .. versionadded:: 8.0
+ formatter_class: t.Type["HelpFormatter"] = HelpFormatter
+
+ def __init__(
+ self,
+ command: "Command",
+ parent: t.Optional["Context"] = None,
+ info_name: t.Optional[str] = None,
+ obj: t.Optional[t.Any] = None,
+ auto_envvar_prefix: t.Optional[str] = None,
+ default_map: t.Optional[t.MutableMapping[str, t.Any]] = None,
+ terminal_width: t.Optional[int] = None,
+ max_content_width: t.Optional[int] = None,
+ resilient_parsing: bool = False,
+ allow_extra_args: t.Optional[bool] = None,
+ allow_interspersed_args: t.Optional[bool] = None,
+ ignore_unknown_options: t.Optional[bool] = None,
+ help_option_names: t.Optional[t.List[str]] = None,
+ token_normalize_func: t.Optional[t.Callable[[str], str]] = None,
+ color: t.Optional[bool] = None,
+ show_default: t.Optional[bool] = None,
+ ) -> None:
+ #: the parent context or `None` if none exists.
self.parent = parent
+ #: the :class:`Command` for this context.
self.command = command
+ #: the descriptive information name
self.info_name = info_name
+ #: Map of parameter names to their parsed values. Parameters
+ #: with ``expose_value=False`` are not stored.
self.params: t.Dict[str, t.Any] = {}
+ #: the leftover arguments.
self.args: t.List[str] = []
+ #: protected arguments. These are arguments that are prepended
+ #: to `args` when certain parsing scenarios are encountered but
+ #: must be never propagated to another arguments. This is used
+ #: to implement nested parsing.
self.protected_args: t.List[str] = []
- self._opt_prefixes: t.Set[str] = set(parent._opt_prefixes
- ) if parent else set()
+ #: the collected prefixes of the command's options.
+ self._opt_prefixes: t.Set[str] = set(parent._opt_prefixes) if parent else set()
+
if obj is None and parent is not None:
obj = parent.obj
+
+ #: the user object stored.
self.obj: t.Any = obj
- self._meta: t.Dict[str, t.Any] = getattr(parent, 'meta', {})
- if (default_map is None and info_name is not None and parent is not
- None and parent.default_map is not None):
+ self._meta: t.Dict[str, t.Any] = getattr(parent, "meta", {})
+
+ #: A dictionary (-like object) with defaults for parameters.
+ if (
+ default_map is None
+ and info_name is not None
+ and parent is not None
+ and parent.default_map is not None
+ ):
default_map = parent.default_map.get(info_name)
- self.default_map: t.Optional[t.MutableMapping[str, t.Any]
- ] = default_map
+
+ self.default_map: t.Optional[t.MutableMapping[str, t.Any]] = default_map
+
+ #: This flag indicates if a subcommand is going to be executed. A
+ #: group callback can use this information to figure out if it's
+ #: being executed directly or because the execution flow passes
+ #: onwards to a subcommand. By default it's None, but it can be
+ #: the name of the subcommand to execute.
+ #:
+ #: If chaining is enabled this will be set to ``'*'`` in case
+ #: any commands are executed. It is however not possible to
+ #: figure out which ones. If you require this knowledge you
+ #: should use a :func:`result_callback`.
self.invoked_subcommand: t.Optional[str] = None
+
if terminal_width is None and parent is not None:
terminal_width = parent.terminal_width
+
+ #: The width of the terminal (None is autodetection).
self.terminal_width: t.Optional[int] = terminal_width
+
if max_content_width is None and parent is not None:
max_content_width = parent.max_content_width
+
+ #: The maximum width of formatted content (None implies a sensible
+ #: default which is 80 for most things).
self.max_content_width: t.Optional[int] = max_content_width
+
if allow_extra_args is None:
allow_extra_args = command.allow_extra_args
+
+ #: Indicates if the context allows extra args or if it should
+ #: fail on parsing.
+ #:
+ #: .. versionadded:: 3.0
self.allow_extra_args = allow_extra_args
+
if allow_interspersed_args is None:
allow_interspersed_args = command.allow_interspersed_args
+
+ #: Indicates if the context allows mixing of arguments and
+ #: options or not.
+ #:
+ #: .. versionadded:: 3.0
self.allow_interspersed_args: bool = allow_interspersed_args
+
if ignore_unknown_options is None:
ignore_unknown_options = command.ignore_unknown_options
+
+ #: Instructs click to ignore options that a command does not
+ #: understand and will store it on the context for later
+ #: processing. This is primarily useful for situations where you
+ #: want to call into external programs. Generally this pattern is
+ #: strongly discouraged because it's not possibly to losslessly
+ #: forward all arguments.
+ #:
+ #: .. versionadded:: 4.0
self.ignore_unknown_options: bool = ignore_unknown_options
+
if help_option_names is None:
if parent is not None:
help_option_names = parent.help_option_names
else:
- help_option_names = ['--help']
+ help_option_names = ["--help"]
+
+ #: The names for the help options.
self.help_option_names: t.List[str] = help_option_names
+
if token_normalize_func is None and parent is not None:
token_normalize_func = parent.token_normalize_func
- self.token_normalize_func: t.Optional[t.Callable[[str], str]
- ] = token_normalize_func
+
+ #: An optional normalization function for tokens. This is
+ #: options, choices, commands etc.
+ self.token_normalize_func: t.Optional[
+ t.Callable[[str], str]
+ ] = token_normalize_func
+
+ #: Indicates if resilient parsing is enabled. In that case Click
+ #: will do its best to not cause any failures and default values
+ #: will be ignored. Useful for completion.
self.resilient_parsing: bool = resilient_parsing
+
+ # If there is no envvar prefix yet, but the parent has one and
+ # the command on this level has a name, we can expand the envvar
+ # prefix automatically.
if auto_envvar_prefix is None:
- if (parent is not None and parent.auto_envvar_prefix is not
- None and self.info_name is not None):
+ if (
+ parent is not None
+ and parent.auto_envvar_prefix is not None
+ and self.info_name is not None
+ ):
auto_envvar_prefix = (
- f'{parent.auto_envvar_prefix}_{self.info_name.upper()}')
+ f"{parent.auto_envvar_prefix}_{self.info_name.upper()}"
+ )
else:
auto_envvar_prefix = auto_envvar_prefix.upper()
+
if auto_envvar_prefix is not None:
- auto_envvar_prefix = auto_envvar_prefix.replace('-', '_')
+ auto_envvar_prefix = auto_envvar_prefix.replace("-", "_")
+
self.auto_envvar_prefix: t.Optional[str] = auto_envvar_prefix
+
if color is None and parent is not None:
color = parent.color
+
+ #: Controls if styling output is wanted or not.
self.color: t.Optional[bool] = color
+
if show_default is None and parent is not None:
show_default = parent.show_default
+
+ #: Show option default values when formatting help text.
self.show_default: t.Optional[bool] = show_default
+
self._close_callbacks: t.List[t.Callable[[], t.Any]] = []
self._depth = 0
self._parameter_source: t.Dict[str, ParameterSource] = {}
self._exit_stack = ExitStack()
- def to_info_dict(self) ->t.Dict[str, t.Any]:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
"""Gather information that could be useful for a tool generating
user-facing documentation. This traverses the entire CLI
structure.
@@ -278,23 +441,33 @@ class Context:
.. versionadded:: 8.0
"""
- pass
-
- def __enter__(self) ->'Context':
+ return {
+ "command": self.command.to_info_dict(self),
+ "info_name": self.info_name,
+ "allow_extra_args": self.allow_extra_args,
+ "allow_interspersed_args": self.allow_interspersed_args,
+ "ignore_unknown_options": self.ignore_unknown_options,
+ "auto_envvar_prefix": self.auto_envvar_prefix,
+ }
+
+ def __enter__(self) -> "Context":
self._depth += 1
push_context(self)
return self
- def __exit__(self, exc_type: t.Optional[t.Type[BaseException]],
- exc_value: t.Optional[BaseException], tb: t.Optional[TracebackType]
- ) ->None:
+ def __exit__(
+ self,
+ exc_type: t.Optional[t.Type[BaseException]],
+ exc_value: t.Optional[BaseException],
+ tb: t.Optional[TracebackType],
+ ) -> None:
self._depth -= 1
if self._depth == 0:
self.close()
pop_context()
@contextmanager
- def scope(self, cleanup: bool=True) ->t.Iterator['Context']:
+ def scope(self, cleanup: bool = True) -> t.Iterator["Context"]:
"""This helper method can be used with the context object to promote
it to the current thread local (see :func:`get_current_context`).
The default behavior of this is to invoke the cleanup functions which
@@ -322,10 +495,17 @@ class Context:
temporarily pushed in which case this can be disabled.
Nested pushes automatically defer the cleanup.
"""
- pass
+ if not cleanup:
+ self._depth += 1
+ try:
+ with self as rv:
+ yield rv
+ finally:
+ if not cleanup:
+ self._depth -= 1
@property
- def meta(self) ->t.Dict[str, t.Any]:
+ def meta(self) -> t.Dict[str, t.Any]:
"""This is a dictionary which is shared with all the contexts
that are nested. It exists so that click utilities can store some
state here if they need to. It is however the responsibility of
@@ -350,9 +530,9 @@ class Context:
.. versionadded:: 5.0
"""
- pass
+ return self._meta
- def make_formatter(self) ->HelpFormatter:
+ def make_formatter(self) -> HelpFormatter:
"""Creates the :class:`~click.HelpFormatter` for the help and
usage output.
@@ -362,9 +542,11 @@ class Context:
.. versionchanged:: 8.0
Added the :attr:`formatter_class` attribute.
"""
- pass
+ return self.formatter_class(
+ width=self.terminal_width, max_width=self.max_content_width
+ )
- def with_resource(self, context_manager: t.ContextManager[V]) ->V:
+ def with_resource(self, context_manager: t.ContextManager[V]) -> V:
"""Register a resource as if it were used in a ``with``
statement. The resource will be cleaned up when the context is
popped.
@@ -391,10 +573,9 @@ class Context:
.. versionadded:: 8.0
"""
- pass
+ return self._exit_stack.enter_context(context_manager)
- def call_on_close(self, f: t.Callable[..., t.Any]) ->t.Callable[..., t.Any
- ]:
+ def call_on_close(self, f: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]:
"""Register a function to be called when the context tears down.
This can be used to close resources opened during the script
@@ -404,38 +585,77 @@ class Context:
:param f: The function to execute on teardown.
"""
- pass
+ return self._exit_stack.callback(f)
- def close(self) ->None:
+ def close(self) -> None:
"""Invoke all close callbacks registered with
:meth:`call_on_close`, and exit all context managers entered
with :meth:`with_resource`.
"""
- pass
+ self._exit_stack.close()
+ # In case the context is reused, create a new exit stack.
+ self._exit_stack = ExitStack()
@property
- def command_path(self) ->str:
+ def command_path(self) -> str:
"""The computed command path. This is used for the ``usage``
information on the help page. It's automatically created by
combining the info names of the chain of contexts to the root.
"""
- pass
+ rv = ""
+ if self.info_name is not None:
+ rv = self.info_name
+ if self.parent is not None:
+ parent_command_path = [self.parent.command_path]
+
+ if isinstance(self.parent.command, Command):
+ for param in self.parent.command.get_params(self):
+ parent_command_path.extend(param.get_usage_pieces(self))
- def find_root(self) ->'Context':
+ rv = f"{' '.join(parent_command_path)} {rv}"
+ return rv.lstrip()
+
+ def find_root(self) -> "Context":
"""Finds the outermost context."""
- pass
+ node = self
+ while node.parent is not None:
+ node = node.parent
+ return node
- def find_object(self, object_type: t.Type[V]) ->t.Optional[V]:
+ def find_object(self, object_type: t.Type[V]) -> t.Optional[V]:
"""Finds the closest object of a given type."""
- pass
+ node: t.Optional["Context"] = self
- def ensure_object(self, object_type: t.Type[V]) ->V:
+ while node is not None:
+ if isinstance(node.obj, object_type):
+ return node.obj
+
+ node = node.parent
+
+ return None
+
+ def ensure_object(self, object_type: t.Type[V]) -> V:
"""Like :meth:`find_object` but sets the innermost object to a
new instance of `object_type` if it does not exist.
"""
- pass
-
- def lookup_default(self, name: str, call: bool=True) ->t.Optional[t.Any]:
+ rv = self.find_object(object_type)
+ if rv is None:
+ self.obj = rv = object_type()
+ return rv
+
+ @t.overload
+ def lookup_default(
+ self, name: str, call: "te.Literal[True]" = True
+ ) -> t.Optional[t.Any]:
+ ...
+
+ @t.overload
+ def lookup_default(
+ self, name: str, call: "te.Literal[False]" = ...
+ ) -> t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]:
+ ...
+
+ def lookup_default(self, name: str, call: bool = True) -> t.Optional[t.Any]:
"""Get the default for a parameter from :attr:`default_map`.
:param name: Name of the parameter.
@@ -445,46 +665,76 @@ class Context:
.. versionchanged:: 8.0
Added the ``call`` parameter.
"""
- pass
+ if self.default_map is not None:
+ value = self.default_map.get(name)
+
+ if call and callable(value):
+ return value()
+
+ return value
- def fail(self, message: str) ->'te.NoReturn':
+ return None
+
+ def fail(self, message: str) -> "te.NoReturn":
"""Aborts the execution of the program with a specific error
message.
:param message: the error message to fail with.
"""
- pass
+ raise UsageError(message, self)
- def abort(self) ->'te.NoReturn':
+ def abort(self) -> "te.NoReturn":
"""Aborts the script."""
- pass
+ raise Abort()
- def exit(self, code: int=0) ->'te.NoReturn':
+ def exit(self, code: int = 0) -> "te.NoReturn":
"""Exits the application with a given exit code."""
- pass
+ raise Exit(code)
- def get_usage(self) ->str:
+ def get_usage(self) -> str:
"""Helper method to get formatted usage string for the current
context and command.
"""
- pass
+ return self.command.get_usage(self)
- def get_help(self) ->str:
+ def get_help(self) -> str:
"""Helper method to get formatted help page for the current
context and command.
"""
- pass
+ return self.command.get_help(self)
- def _make_sub_context(self, command: 'Command') ->'Context':
+ def _make_sub_context(self, command: "Command") -> "Context":
"""Create a new context of the same type as this context, but
for a new command.
:meta private:
"""
- pass
-
- def invoke(__self, __callback: t.Union['Command', 't.Callable[..., V]'],
- *args: t.Any, **kwargs: t.Any) ->t.Union[t.Any, V]:
+ return type(self)(command, info_name=command.name, parent=self)
+
+ @t.overload
+ def invoke(
+ __self, # noqa: B902
+ __callback: "t.Callable[..., V]",
+ *args: t.Any,
+ **kwargs: t.Any,
+ ) -> V:
+ ...
+
+ @t.overload
+ def invoke(
+ __self, # noqa: B902
+ __callback: "Command",
+ *args: t.Any,
+ **kwargs: t.Any,
+ ) -> t.Any:
+ ...
+
+ def invoke(
+ __self, # noqa: B902
+ __callback: t.Union["Command", "t.Callable[..., V]"],
+ *args: t.Any,
+ **kwargs: t.Any,
+ ) -> t.Union[t.Any, V]:
"""Invokes a command callback in exactly the way it expects. There
are two ways to invoke this method:
@@ -504,10 +754,37 @@ class Context:
All ``kwargs`` are tracked in :attr:`params` so they will be
passed if :meth:`forward` is called at multiple levels.
"""
- pass
+ if isinstance(__callback, Command):
+ other_cmd = __callback
+
+ if other_cmd.callback is None:
+ raise TypeError(
+ "The given command does not have a callback that can be invoked."
+ )
+ else:
+ __callback = t.cast("t.Callable[..., V]", other_cmd.callback)
+
+ ctx = __self._make_sub_context(other_cmd)
+
+ for param in other_cmd.params:
+ if param.name not in kwargs and param.expose_value:
+ kwargs[param.name] = param.type_cast_value( # type: ignore
+ ctx, param.get_default(ctx)
+ )
+
+ # Track all kwargs as params, so that forward() will pass
+ # them on in subsequent calls.
+ ctx.params.update(kwargs)
+ else:
+ ctx = __self
+
+ with augment_usage_errors(__self):
+ with ctx:
+ return __callback(*args, **kwargs)
- def forward(__self, __cmd: 'Command', *args: t.Any, **kwargs: t.Any
- ) ->t.Any:
+ def forward(
+ __self, __cmd: "Command", *args: t.Any, **kwargs: t.Any # noqa: B902
+ ) -> t.Any:
"""Similar to :meth:`invoke` but fills in default keyword
arguments from the current context if the other command expects
it. This cannot invoke callbacks directly, only other commands.
@@ -516,18 +793,26 @@ class Context:
All ``kwargs`` are tracked in :attr:`params` so they will be
passed if ``forward`` is called at multiple levels.
"""
- pass
+ # Can only forward to other commands, not direct callbacks.
+ if not isinstance(__cmd, Command):
+ raise TypeError("Callback is not a command.")
+
+ for param in __self.params:
+ if param not in kwargs:
+ kwargs[param] = __self.params[param]
+
+ return __self.invoke(__cmd, *args, **kwargs)
- def set_parameter_source(self, name: str, source: ParameterSource) ->None:
+ def set_parameter_source(self, name: str, source: ParameterSource) -> None:
"""Set the source of a parameter. This indicates the location
from which the value of the parameter was obtained.
:param name: The name of the parameter.
:param source: A member of :class:`~click.core.ParameterSource`.
"""
- pass
+ self._parameter_source[name] = source
- def get_parameter_source(self, name: str) ->t.Optional[ParameterSource]:
+ def get_parameter_source(self, name: str) -> t.Optional[ParameterSource]:
"""Get the source of a parameter. This indicates the location
from which the value of the parameter was obtained.
@@ -543,7 +828,7 @@ class Context:
Returns ``None`` if the parameter was not provided from any
source.
"""
- pass
+ return self._parameter_source.get(name)
class BaseCommand:
@@ -567,19 +852,36 @@ class BaseCommand:
:param context_settings: an optional dictionary with defaults that are
passed to the context object.
"""
+
+ #: The context class to create with :meth:`make_context`.
+ #:
+ #: .. versionadded:: 8.0
context_class: t.Type[Context] = Context
+ #: the default for the :attr:`Context.allow_extra_args` flag.
allow_extra_args = False
+ #: the default for the :attr:`Context.allow_interspersed_args` flag.
allow_interspersed_args = True
+ #: the default for the :attr:`Context.ignore_unknown_options` flag.
ignore_unknown_options = False
- def __init__(self, name: t.Optional[str], context_settings: t.Optional[
- t.MutableMapping[str, t.Any]]=None) ->None:
+ def __init__(
+ self,
+ name: t.Optional[str],
+ context_settings: t.Optional[t.MutableMapping[str, t.Any]] = None,
+ ) -> None:
+ #: the name the command thinks it has. Upon registering a command
+ #: on a :class:`Group` the group will default the command name
+ #: with this information. You should instead use the
+ #: :class:`Context`\'s :attr:`~Context.info_name` attribute.
self.name = name
+
if context_settings is None:
context_settings = {}
+
+ #: an optional dictionary with defaults passed to the context.
self.context_settings: t.MutableMapping[str, t.Any] = context_settings
- def to_info_dict(self, ctx: Context) ->t.Dict[str, t.Any]:
+ def to_info_dict(self, ctx: Context) -> t.Dict[str, t.Any]:
"""Gather information that could be useful for a tool generating
user-facing documentation. This traverses the entire structure
below this command.
@@ -591,13 +893,24 @@ class BaseCommand:
.. versionadded:: 8.0
"""
- pass
+ return {"name": self.name}
+
+ def __repr__(self) -> str:
+ return f"<{self.__class__.__name__} {self.name}>"
- def __repr__(self) ->str:
- return f'<{self.__class__.__name__} {self.name}>'
+ def get_usage(self, ctx: Context) -> str:
+ raise NotImplementedError("Base commands cannot get usage")
- def make_context(self, info_name: t.Optional[str], args: t.List[str],
- parent: t.Optional[Context]=None, **extra: t.Any) ->Context:
+ def get_help(self, ctx: Context) -> str:
+ raise NotImplementedError("Base commands cannot get help")
+
+ def make_context(
+ self,
+ info_name: t.Optional[str],
+ args: t.List[str],
+ parent: t.Optional[Context] = None,
+ **extra: t.Any,
+ ) -> Context:
"""This function when given an info name and arguments will kick
off the parsing and create a new :class:`Context`. It does not
invoke the actual command callback though.
@@ -618,23 +931,32 @@ class BaseCommand:
.. versionchanged:: 8.0
Added the :attr:`context_class` attribute.
"""
- pass
+ for key, value in self.context_settings.items():
+ if key not in extra:
+ extra[key] = value
+
+ ctx = self.context_class(
+ self, info_name=info_name, parent=parent, **extra # type: ignore
+ )
- def parse_args(self, ctx: Context, args: t.List[str]) ->t.List[str]:
+ with ctx.scope(cleanup=False):
+ self.parse_args(ctx, args)
+ return ctx
+
+ def parse_args(self, ctx: Context, args: t.List[str]) -> t.List[str]:
"""Given a context and a list of arguments this creates the parser
and parses the arguments, then modifies the context as necessary.
This is automatically invoked by :meth:`make_context`.
"""
- pass
+ raise NotImplementedError("Base commands do not know how to parse arguments.")
- def invoke(self, ctx: Context) ->t.Any:
+ def invoke(self, ctx: Context) -> t.Any:
"""Given a context, this invokes the command. The default
implementation is raising a not implemented error.
"""
- pass
+ raise NotImplementedError("Base commands are not invocable by default")
- def shell_complete(self, ctx: Context, incomplete: str) ->t.List[
- 'CompletionItem']:
+ def shell_complete(self, ctx: Context, incomplete: str) -> t.List["CompletionItem"]:
"""Return a list of completions for the incomplete value. Looks
at the names of chained multi-commands.
@@ -647,12 +969,53 @@ class BaseCommand:
.. versionadded:: 8.0
"""
- pass
-
- def main(self, args: t.Optional[t.Sequence[str]]=None, prog_name: t.
- Optional[str]=None, complete_var: t.Optional[str]=None,
- standalone_mode: bool=True, windows_expand_args: bool=True, **extra:
- t.Any) ->t.Any:
+ from click.shell_completion import CompletionItem
+
+ results: t.List["CompletionItem"] = []
+
+ while ctx.parent is not None:
+ ctx = ctx.parent
+
+ if isinstance(ctx.command, MultiCommand) and ctx.command.chain:
+ results.extend(
+ CompletionItem(name, help=command.get_short_help_str())
+ for name, command in _complete_visible_commands(ctx, incomplete)
+ if name not in ctx.protected_args
+ )
+
+ return results
+
+ @t.overload
+ def main(
+ self,
+ args: t.Optional[t.Sequence[str]] = None,
+ prog_name: t.Optional[str] = None,
+ complete_var: t.Optional[str] = None,
+ standalone_mode: "te.Literal[True]" = True,
+ **extra: t.Any,
+ ) -> "te.NoReturn":
+ ...
+
+ @t.overload
+ def main(
+ self,
+ args: t.Optional[t.Sequence[str]] = None,
+ prog_name: t.Optional[str] = None,
+ complete_var: t.Optional[str] = None,
+ standalone_mode: bool = ...,
+ **extra: t.Any,
+ ) -> t.Any:
+ ...
+
+ def main(
+ self,
+ args: t.Optional[t.Sequence[str]] = None,
+ prog_name: t.Optional[str] = None,
+ complete_var: t.Optional[str] = None,
+ standalone_mode: bool = True,
+ windows_expand_args: bool = True,
+ **extra: t.Any,
+ ) -> t.Any:
"""This is the way to invoke a script with all the bells and
whistles as a command line application. This will always terminate
the application after a call. If this is not wanted, ``SystemExit``
@@ -695,10 +1058,74 @@ class BaseCommand:
.. versionchanged:: 3.0
Added the ``standalone_mode`` parameter.
"""
- pass
+ if args is None:
+ args = sys.argv[1:]
- def _main_shell_completion(self, ctx_args: t.MutableMapping[str, t.Any],
- prog_name: str, complete_var: t.Optional[str]=None) ->None:
+ if os.name == "nt" and windows_expand_args:
+ args = _expand_args(args)
+ else:
+ args = list(args)
+
+ if prog_name is None:
+ prog_name = _detect_program_name()
+
+ # Process shell completion requests and exit early.
+ self._main_shell_completion(extra, prog_name, complete_var)
+
+ try:
+ try:
+ with self.make_context(prog_name, args, **extra) as ctx:
+ rv = self.invoke(ctx)
+ if not standalone_mode:
+ return rv
+ # it's not safe to `ctx.exit(rv)` here!
+ # note that `rv` may actually contain data like "1" which
+ # has obvious effects
+ # more subtle case: `rv=[None, None]` can come out of
+ # chained commands which all returned `None` -- so it's not
+ # even always obvious that `rv` indicates success/failure
+ # by its truthiness/falsiness
+ ctx.exit()
+ except (EOFError, KeyboardInterrupt) as e:
+ echo(file=sys.stderr)
+ raise Abort() from e
+ except ClickException as e:
+ if not standalone_mode:
+ raise
+ e.show()
+ sys.exit(e.exit_code)
+ except OSError as e:
+ if e.errno == errno.EPIPE:
+ sys.stdout = t.cast(t.TextIO, PacifyFlushWrapper(sys.stdout))
+ sys.stderr = t.cast(t.TextIO, PacifyFlushWrapper(sys.stderr))
+ sys.exit(1)
+ else:
+ raise
+ except Exit as e:
+ if standalone_mode:
+ sys.exit(e.exit_code)
+ else:
+ # in non-standalone mode, return the exit code
+ # note that this is only reached if `self.invoke` above raises
+ # an Exit explicitly -- thus bypassing the check there which
+ # would return its result
+ # the results of non-standalone execution may therefore be
+ # somewhat ambiguous: if there are codepaths which lead to
+ # `ctx.exit(1)` and to `return 1`, the caller won't be able to
+ # tell the difference between the two
+ return e.exit_code
+ except Abort:
+ if not standalone_mode:
+ raise
+ echo(_("Aborted!"), file=sys.stderr)
+ sys.exit(1)
+
+ def _main_shell_completion(
+ self,
+ ctx_args: t.MutableMapping[str, t.Any],
+ prog_name: str,
+ complete_var: t.Optional[str] = None,
+ ) -> None:
"""Check if the shell is asking for tab completion, process
that, then exit early. Called from :meth:`main` before the
program is invoked.
@@ -711,9 +1138,21 @@ class BaseCommand:
.. versionchanged:: 8.2.0
Dots (``.``) in ``prog_name`` are replaced with underscores (``_``).
"""
- pass
+ if complete_var is None:
+ complete_name = prog_name.replace("-", "_").replace(".", "_")
+ complete_var = f"_{complete_name}_COMPLETE".upper()
+
+ instruction = os.environ.get(complete_var)
+
+ if not instruction:
+ return
- def __call__(self, *args: t.Any, **kwargs: t.Any) ->t.Any:
+ from .shell_completion import shell_complete
+
+ rv = shell_complete(self, ctx_args, prog_name, complete_var, instruction)
+ sys.exit(rv)
+
+ def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
"""Alias for :meth:`main`."""
return self.main(*args, **kwargs)
@@ -760,16 +1199,29 @@ class Command(BaseCommand):
Added the ``context_settings`` parameter.
"""
- def __init__(self, name: t.Optional[str], context_settings: t.Optional[
- t.MutableMapping[str, t.Any]]=None, callback: t.Optional[t.Callable
- [..., t.Any]]=None, params: t.Optional[t.List['Parameter']]=None,
- help: t.Optional[str]=None, epilog: t.Optional[str]=None,
- short_help: t.Optional[str]=None, options_metavar: t.Optional[str]=
- '[OPTIONS]', add_help_option: bool=True, no_args_is_help: bool=
- False, hidden: bool=False, deprecated: bool=False) ->None:
+ def __init__(
+ self,
+ name: t.Optional[str],
+ context_settings: t.Optional[t.MutableMapping[str, t.Any]] = None,
+ callback: t.Optional[t.Callable[..., t.Any]] = None,
+ params: t.Optional[t.List["Parameter"]] = None,
+ help: t.Optional[str] = None,
+ epilog: t.Optional[str] = None,
+ short_help: t.Optional[str] = None,
+ options_metavar: t.Optional[str] = "[OPTIONS]",
+ add_help_option: bool = True,
+ no_args_is_help: bool = False,
+ hidden: bool = False,
+ deprecated: bool = False,
+ ) -> None:
super().__init__(name, context_settings)
+ #: the callback to execute when the command fires. This might be
+ #: `None` in which case nothing happens.
self.callback = callback
- self.params: t.List['Parameter'] = params or []
+ #: the list of parameters for this command in the order they
+ #: should show up in the help page and execute. Eager parameters
+ #: will automatically be handled before non eager ones.
+ self.params: t.List["Parameter"] = params or []
self.help = help
self.epilog = epilog
self.options_metavar = options_metavar
@@ -779,52 +1231,117 @@ class Command(BaseCommand):
self.hidden = hidden
self.deprecated = deprecated
- def get_usage(self, ctx: Context) ->str:
+ def to_info_dict(self, ctx: Context) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict(ctx)
+ info_dict.update(
+ params=[param.to_info_dict() for param in self.get_params(ctx)],
+ help=self.help,
+ epilog=self.epilog,
+ short_help=self.short_help,
+ hidden=self.hidden,
+ deprecated=self.deprecated,
+ )
+ return info_dict
+
+ def get_usage(self, ctx: Context) -> str:
"""Formats the usage line into a string and returns it.
Calls :meth:`format_usage` internally.
"""
- pass
+ formatter = ctx.make_formatter()
+ self.format_usage(ctx, formatter)
+ return formatter.getvalue().rstrip("\n")
- def format_usage(self, ctx: Context, formatter: HelpFormatter) ->None:
+ def get_params(self, ctx: Context) -> t.List["Parameter"]:
+ rv = self.params
+ help_option = self.get_help_option(ctx)
+
+ if help_option is not None:
+ rv = [*rv, help_option]
+
+ return rv
+
+ def format_usage(self, ctx: Context, formatter: HelpFormatter) -> None:
"""Writes the usage line into the formatter.
This is a low-level method called by :meth:`get_usage`.
"""
- pass
+ pieces = self.collect_usage_pieces(ctx)
+ formatter.write_usage(ctx.command_path, " ".join(pieces))
- def collect_usage_pieces(self, ctx: Context) ->t.List[str]:
+ def collect_usage_pieces(self, ctx: Context) -> t.List[str]:
"""Returns all the pieces that go into the usage line and returns
it as a list of strings.
"""
- pass
+ rv = [self.options_metavar] if self.options_metavar else []
- def get_help_option_names(self, ctx: Context) ->t.List[str]:
+ for param in self.get_params(ctx):
+ rv.extend(param.get_usage_pieces(ctx))
+
+ return rv
+
+ def get_help_option_names(self, ctx: Context) -> t.List[str]:
"""Returns the names for the help option."""
- pass
+ all_names = set(ctx.help_option_names)
+ for param in self.params:
+ all_names.difference_update(param.opts)
+ all_names.difference_update(param.secondary_opts)
+ return list(all_names)
- def get_help_option(self, ctx: Context) ->t.Optional['Option']:
+ def get_help_option(self, ctx: Context) -> t.Optional["Option"]:
"""Returns the help option object."""
- pass
-
- def make_parser(self, ctx: Context) ->OptionParser:
+ help_options = self.get_help_option_names(ctx)
+
+ if not help_options or not self.add_help_option:
+ return None
+
+ def show_help(ctx: Context, param: "Parameter", value: str) -> None:
+ if value and not ctx.resilient_parsing:
+ echo(ctx.get_help(), color=ctx.color)
+ ctx.exit()
+
+ return Option(
+ help_options,
+ is_flag=True,
+ is_eager=True,
+ expose_value=False,
+ callback=show_help,
+ help=_("Show this message and exit."),
+ )
+
+ def make_parser(self, ctx: Context) -> OptionParser:
"""Creates the underlying option parser for this command."""
- pass
+ parser = OptionParser(ctx)
+ for param in self.get_params(ctx):
+ param.add_to_parser(parser, ctx)
+ return parser
- def get_help(self, ctx: Context) ->str:
+ def get_help(self, ctx: Context) -> str:
"""Formats the help into a string and returns it.
Calls :meth:`format_help` internally.
"""
- pass
+ formatter = ctx.make_formatter()
+ self.format_help(ctx, formatter)
+ return formatter.getvalue().rstrip("\n")
- def get_short_help_str(self, limit: int=45) ->str:
+ def get_short_help_str(self, limit: int = 45) -> str:
"""Gets short help for the command or makes it by shortening the
long help string.
"""
- pass
+ if self.short_help:
+ text = inspect.cleandoc(self.short_help)
+ elif self.help:
+ text = make_default_short_help(self.help, limit)
+ else:
+ text = ""
+
+ if self.deprecated:
+ text = _("(Deprecated) {text}").format(text=text)
+
+ return text.strip()
- def format_help(self, ctx: Context, formatter: HelpFormatter) ->None:
+ def format_help(self, ctx: Context, formatter: HelpFormatter) -> None:
"""Writes the help into the formatter if it exists.
This is a low-level method called by :meth:`get_help`.
@@ -836,28 +1353,87 @@ class Command(BaseCommand):
- :meth:`format_options`
- :meth:`format_epilog`
"""
- pass
+ self.format_usage(ctx, formatter)
+ self.format_help_text(ctx, formatter)
+ self.format_options(ctx, formatter)
+ self.format_epilog(ctx, formatter)
- def format_help_text(self, ctx: Context, formatter: HelpFormatter) ->None:
+ def format_help_text(self, ctx: Context, formatter: HelpFormatter) -> None:
"""Writes the help text to the formatter if it exists."""
- pass
+ if self.help is not None:
+ # truncate the help text to the first form feed
+ text = inspect.cleandoc(self.help).partition("\f")[0]
+ else:
+ text = ""
+
+ if self.deprecated:
+ text = _("(Deprecated) {text}").format(text=text)
+
+ if text:
+ formatter.write_paragraph()
+
+ with formatter.indentation():
+ formatter.write_text(text)
- def format_options(self, ctx: Context, formatter: HelpFormatter) ->None:
+ def format_options(self, ctx: Context, formatter: HelpFormatter) -> None:
"""Writes all the options into the formatter if they exist."""
- pass
+ opts = []
+ for param in self.get_params(ctx):
+ rv = param.get_help_record(ctx)
+ if rv is not None:
+ opts.append(rv)
- def format_epilog(self, ctx: Context, formatter: HelpFormatter) ->None:
- """Writes the epilog into the formatter if it exists."""
- pass
+ if opts:
+ with formatter.section(_("Options")):
+ formatter.write_dl(opts)
- def invoke(self, ctx: Context) ->t.Any:
+ def format_epilog(self, ctx: Context, formatter: HelpFormatter) -> None:
+ """Writes the epilog into the formatter if it exists."""
+ if self.epilog:
+ epilog = inspect.cleandoc(self.epilog)
+ formatter.write_paragraph()
+
+ with formatter.indentation():
+ formatter.write_text(epilog)
+
+ def parse_args(self, ctx: Context, args: t.List[str]) -> t.List[str]:
+ if not args and self.no_args_is_help and not ctx.resilient_parsing:
+ echo(ctx.get_help(), color=ctx.color)
+ ctx.exit()
+
+ parser = self.make_parser(ctx)
+ opts, args, param_order = parser.parse_args(args=args)
+
+ for param in iter_params_for_processing(param_order, self.get_params(ctx)):
+ value, args = param.handle_parse_result(ctx, opts, args)
+
+ if args and not ctx.allow_extra_args and not ctx.resilient_parsing:
+ ctx.fail(
+ ngettext(
+ "Got unexpected extra argument ({args})",
+ "Got unexpected extra arguments ({args})",
+ len(args),
+ ).format(args=" ".join(map(str, args)))
+ )
+
+ ctx.args = args
+ ctx._opt_prefixes.update(parser._opt_prefixes)
+ return args
+
+ def invoke(self, ctx: Context) -> t.Any:
"""Given a context, this invokes the attached callback (if it exists)
in the right way.
"""
- pass
+ if self.deprecated:
+ message = _(
+ "DeprecationWarning: The command {name!r} is deprecated."
+ ).format(name=self.name)
+ echo(style(message, fg="red"), err=True)
- def shell_complete(self, ctx: Context, incomplete: str) ->t.List[
- 'CompletionItem']:
+ if self.callback is not None:
+ return ctx.invoke(self.callback, **ctx.params)
+
+ def shell_complete(self, ctx: Context, incomplete: str) -> t.List["CompletionItem"]:
"""Return a list of completions for the incomplete value. Looks
at the names of options and chained multi-commands.
@@ -866,7 +1442,31 @@ class Command(BaseCommand):
.. versionadded:: 8.0
"""
- pass
+ from click.shell_completion import CompletionItem
+
+ results: t.List["CompletionItem"] = []
+
+ if incomplete and not incomplete[0].isalnum():
+ for param in self.get_params(ctx):
+ if (
+ not isinstance(param, Option)
+ or param.hidden
+ or (
+ not param.multiple
+ and ctx.get_parameter_source(param.name) # type: ignore
+ is ParameterSource.COMMANDLINE
+ )
+ ):
+ continue
+
+ results.extend(
+ CompletionItem(name, help=param.help)
+ for name in [*param.opts, *param.secondary_opts]
+ if name.startswith(incomplete)
+ )
+
+ results.extend(super().shell_complete(ctx, incomplete))
+ return results
class MultiCommand(Command):
@@ -894,36 +1494,76 @@ class MultiCommand(Command):
:meth:`result_callback` decorator.
:param attrs: Other command arguments described in :class:`Command`.
"""
+
allow_extra_args = True
allow_interspersed_args = False
- def __init__(self, name: t.Optional[str]=None, invoke_without_command:
- bool=False, no_args_is_help: t.Optional[bool]=None,
- subcommand_metavar: t.Optional[str]=None, chain: bool=False,
- result_callback: t.Optional[t.Callable[..., t.Any]]=None, **attrs:
- t.Any) ->None:
+ def __init__(
+ self,
+ name: t.Optional[str] = None,
+ invoke_without_command: bool = False,
+ no_args_is_help: t.Optional[bool] = None,
+ subcommand_metavar: t.Optional[str] = None,
+ chain: bool = False,
+ result_callback: t.Optional[t.Callable[..., t.Any]] = None,
+ **attrs: t.Any,
+ ) -> None:
super().__init__(name, **attrs)
+
if no_args_is_help is None:
no_args_is_help = not invoke_without_command
+
self.no_args_is_help = no_args_is_help
self.invoke_without_command = invoke_without_command
+
if subcommand_metavar is None:
if chain:
- subcommand_metavar = (
- 'COMMAND1 [ARGS]... [COMMAND2 [ARGS]...]...')
+ subcommand_metavar = "COMMAND1 [ARGS]... [COMMAND2 [ARGS]...]..."
else:
- subcommand_metavar = 'COMMAND [ARGS]...'
+ subcommand_metavar = "COMMAND [ARGS]..."
+
self.subcommand_metavar = subcommand_metavar
self.chain = chain
+ # The result callback that is stored. This can be set or
+ # overridden with the :func:`result_callback` decorator.
self._result_callback = result_callback
+
if self.chain:
for param in self.params:
if isinstance(param, Argument) and not param.required:
raise RuntimeError(
- 'Multi commands in chain mode cannot have optional arguments.'
- )
+ "Multi commands in chain mode cannot have"
+ " optional arguments."
+ )
+
+ def to_info_dict(self, ctx: Context) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict(ctx)
+ commands = {}
+
+ for name in self.list_commands(ctx):
+ command = self.get_command(ctx, name)
+
+ if command is None:
+ continue
+
+ sub_ctx = ctx._make_sub_context(command)
- def result_callback(self, replace: bool=False) ->t.Callable[[F], F]:
+ with sub_ctx.scope(cleanup=False):
+ commands[name] = command.to_info_dict(sub_ctx)
+
+ info_dict.update(commands=commands, chain=self.chain)
+ return info_dict
+
+ def collect_usage_pieces(self, ctx: Context) -> t.List[str]:
+ rv = super().collect_usage_pieces(ctx)
+ rv.append(self.subcommand_metavar)
+ return rv
+
+ def format_options(self, ctx: Context, formatter: HelpFormatter) -> None:
+ super().format_options(ctx, formatter)
+ self.format_commands(ctx, formatter)
+
+ def result_callback(self, replace: bool = False) -> t.Callable[[F], F]:
"""Adds a result callback to the command. By default if a
result callback is already registered this will chain them but
this can be disabled with the `replace` parameter. The result
@@ -951,28 +1591,174 @@ class MultiCommand(Command):
.. versionadded:: 3.0
"""
- pass
- def format_commands(self, ctx: Context, formatter: HelpFormatter) ->None:
+ def decorator(f: F) -> F:
+ old_callback = self._result_callback
+
+ if old_callback is None or replace:
+ self._result_callback = f
+ return f
+
+ def function(__value, *args, **kwargs): # type: ignore
+ inner = old_callback(__value, *args, **kwargs)
+ return f(inner, *args, **kwargs)
+
+ self._result_callback = rv = update_wrapper(t.cast(F, function), f)
+ return rv
+
+ return decorator
+
+ def format_commands(self, ctx: Context, formatter: HelpFormatter) -> None:
"""Extra format methods for multi methods that adds all the commands
after the options.
"""
- pass
+ commands = []
+ for subcommand in self.list_commands(ctx):
+ cmd = self.get_command(ctx, subcommand)
+ # What is this, the tool lied about a command. Ignore it
+ if cmd is None:
+ continue
+ if cmd.hidden:
+ continue
+
+ commands.append((subcommand, cmd))
+
+ # allow for 3 times the default spacing
+ if len(commands):
+ limit = formatter.width - 6 - max(len(cmd[0]) for cmd in commands)
+
+ rows = []
+ for subcommand, cmd in commands:
+ help = cmd.get_short_help_str(limit)
+ rows.append((subcommand, help))
+
+ if rows:
+ with formatter.section(_("Commands")):
+ formatter.write_dl(rows)
+
+ def parse_args(self, ctx: Context, args: t.List[str]) -> t.List[str]:
+ if not args and self.no_args_is_help and not ctx.resilient_parsing:
+ echo(ctx.get_help(), color=ctx.color)
+ ctx.exit()
- def get_command(self, ctx: Context, cmd_name: str) ->t.Optional[Command]:
+ rest = super().parse_args(ctx, args)
+
+ if self.chain:
+ ctx.protected_args = rest
+ ctx.args = []
+ elif rest:
+ ctx.protected_args, ctx.args = rest[:1], rest[1:]
+
+ return ctx.args
+
+ def invoke(self, ctx: Context) -> t.Any:
+ def _process_result(value: t.Any) -> t.Any:
+ if self._result_callback is not None:
+ value = ctx.invoke(self._result_callback, value, **ctx.params)
+ return value
+
+ if not ctx.protected_args:
+ if self.invoke_without_command:
+ # No subcommand was invoked, so the result callback is
+ # invoked with the group return value for regular
+ # groups, or an empty list for chained groups.
+ with ctx:
+ rv = super().invoke(ctx)
+ return _process_result([] if self.chain else rv)
+ ctx.fail(_("Missing command."))
+
+ # Fetch args back out
+ args = [*ctx.protected_args, *ctx.args]
+ ctx.args = []
+ ctx.protected_args = []
+
+ # If we're not in chain mode, we only allow the invocation of a
+ # single command but we also inform the current context about the
+ # name of the command to invoke.
+ if not self.chain:
+ # Make sure the context is entered so we do not clean up
+ # resources until the result processor has worked.
+ with ctx:
+ cmd_name, cmd, args = self.resolve_command(ctx, args)
+ assert cmd is not None
+ ctx.invoked_subcommand = cmd_name
+ super().invoke(ctx)
+ sub_ctx = cmd.make_context(cmd_name, args, parent=ctx)
+ with sub_ctx:
+ return _process_result(sub_ctx.command.invoke(sub_ctx))
+
+ # In chain mode we create the contexts step by step, but after the
+ # base command has been invoked. Because at that point we do not
+ # know the subcommands yet, the invoked subcommand attribute is
+ # set to ``*`` to inform the command that subcommands are executed
+ # but nothing else.
+ with ctx:
+ ctx.invoked_subcommand = "*" if args else None
+ super().invoke(ctx)
+
+ # Otherwise we make every single context and invoke them in a
+ # chain. In that case the return value to the result processor
+ # is the list of all invoked subcommand's results.
+ contexts = []
+ while args:
+ cmd_name, cmd, args = self.resolve_command(ctx, args)
+ assert cmd is not None
+ sub_ctx = cmd.make_context(
+ cmd_name,
+ args,
+ parent=ctx,
+ allow_extra_args=True,
+ allow_interspersed_args=False,
+ )
+ contexts.append(sub_ctx)
+ args, sub_ctx.args = sub_ctx.args, []
+
+ rv = []
+ for sub_ctx in contexts:
+ with sub_ctx:
+ rv.append(sub_ctx.command.invoke(sub_ctx))
+ return _process_result(rv)
+
+ def resolve_command(
+ self, ctx: Context, args: t.List[str]
+ ) -> t.Tuple[t.Optional[str], t.Optional[Command], t.List[str]]:
+ cmd_name = make_str(args[0])
+ original_cmd_name = cmd_name
+
+ # Get the command
+ cmd = self.get_command(ctx, cmd_name)
+
+ # If we can't find the command but there is a normalization
+ # function available, we try with that one.
+ if cmd is None and ctx.token_normalize_func is not None:
+ cmd_name = ctx.token_normalize_func(cmd_name)
+ cmd = self.get_command(ctx, cmd_name)
+
+ # If we don't find the command we want to show an error message
+ # to the user that it was not provided. However, there is
+ # something else we should do: if the first argument looks like
+ # an option we want to kick off parsing again for arguments to
+ # resolve things like --help which now should go to the main
+ # place.
+ if cmd is None and not ctx.resilient_parsing:
+ if split_opt(cmd_name)[0]:
+ self.parse_args(ctx, ctx.args)
+ ctx.fail(_("No such command {name!r}.").format(name=original_cmd_name))
+ return cmd_name if cmd else None, cmd, args[1:]
+
+ def get_command(self, ctx: Context, cmd_name: str) -> t.Optional[Command]:
"""Given a context and a command name, this returns a
:class:`Command` object if it exists or returns `None`.
"""
- pass
+ raise NotImplementedError
- def list_commands(self, ctx: Context) ->t.List[str]:
+ def list_commands(self, ctx: Context) -> t.List[str]:
"""Returns a list of subcommand names in the order they should
appear.
"""
- pass
+ return []
- def shell_complete(self, ctx: Context, incomplete: str) ->t.List[
- 'CompletionItem']:
+ def shell_complete(self, ctx: Context, incomplete: str) -> t.List["CompletionItem"]:
"""Return a list of completions for the incomplete value. Looks
at the names of options, subcommands, and chained
multi-commands.
@@ -982,7 +1768,14 @@ class MultiCommand(Command):
.. versionadded:: 8.0
"""
- pass
+ from click.shell_completion import CompletionItem
+
+ results = [
+ CompletionItem(name, help=command.get_short_help_str())
+ for name, command in _complete_visible_commands(ctx, incomplete)
+ ]
+ results.extend(super().shell_complete(ctx, incomplete))
+ return results
class Group(MultiCommand):
@@ -1000,27 +1793,68 @@ class Group(MultiCommand):
.. versionchanged:: 8.0
The ``commands`` argument can be a list of command objects.
"""
+
+ #: If set, this is used by the group's :meth:`command` decorator
+ #: as the default :class:`Command` class. This is useful to make all
+ #: subcommands use a custom command class.
+ #:
+ #: .. versionadded:: 8.0
command_class: t.Optional[t.Type[Command]] = None
- group_class: t.Optional[t.Union[t.Type['Group'], t.Type[type]]] = None
- def __init__(self, name: t.Optional[str]=None, commands: t.Optional[t.
- Union[t.MutableMapping[str, Command], t.Sequence[Command]]]=None,
- **attrs: t.Any) ->None:
+ #: If set, this is used by the group's :meth:`group` decorator
+ #: as the default :class:`Group` class. This is useful to make all
+ #: subgroups use a custom group class.
+ #:
+ #: If set to the special value :class:`type` (literally
+ #: ``group_class = type``), this group's class will be used as the
+ #: default class. This makes a custom group class continue to make
+ #: custom groups.
+ #:
+ #: .. versionadded:: 8.0
+ group_class: t.Optional[t.Union[t.Type["Group"], t.Type[type]]] = None
+ # Literal[type] isn't valid, so use Type[type]
+
+ def __init__(
+ self,
+ name: t.Optional[str] = None,
+ commands: t.Optional[
+ t.Union[t.MutableMapping[str, Command], t.Sequence[Command]]
+ ] = None,
+ **attrs: t.Any,
+ ) -> None:
super().__init__(name, **attrs)
+
if commands is None:
commands = {}
elif isinstance(commands, abc.Sequence):
commands = {c.name: c for c in commands if c.name is not None}
+
+ #: The registered subcommands by their exported names.
self.commands: t.MutableMapping[str, Command] = commands
- def add_command(self, cmd: Command, name: t.Optional[str]=None) ->None:
+ def add_command(self, cmd: Command, name: t.Optional[str] = None) -> None:
"""Registers another :class:`Command` with this group. If the name
is not provided, the name of the command is used.
"""
- pass
-
- def command(self, *args: t.Any, **kwargs: t.Any) ->t.Union[t.Callable[[
- t.Callable[..., t.Any]], Command], Command]:
+ name = name or cmd.name
+ if name is None:
+ raise TypeError("Command has no name.")
+ _check_multicommand(self, name, cmd, register=True)
+ self.commands[name] = cmd
+
+ @t.overload
+ def command(self, __func: t.Callable[..., t.Any]) -> Command:
+ ...
+
+ @t.overload
+ def command(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Callable[[t.Callable[..., t.Any]], Command]:
+ ...
+
+ def command(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Union[t.Callable[[t.Callable[..., t.Any]], Command], Command]:
"""A shortcut decorator for declaring and attaching a command to
the group. This takes the same arguments as :func:`command` and
immediately registers the created command with this group by
@@ -1035,10 +1869,43 @@ class Group(MultiCommand):
.. versionchanged:: 8.0
Added the :attr:`command_class` attribute.
"""
- pass
+ from .decorators import command
+
+ func: t.Optional[t.Callable[..., t.Any]] = None
+
+ if args and callable(args[0]):
+ assert (
+ len(args) == 1 and not kwargs
+ ), "Use 'command(**kwargs)(callable)' to provide arguments."
+ (func,) = args
+ args = ()
+
+ if self.command_class and kwargs.get("cls") is None:
+ kwargs["cls"] = self.command_class
+
+ def decorator(f: t.Callable[..., t.Any]) -> Command:
+ cmd: Command = command(*args, **kwargs)(f)
+ self.add_command(cmd)
+ return cmd
+
+ if func is not None:
+ return decorator(func)
+
+ return decorator
+
+ @t.overload
+ def group(self, __func: t.Callable[..., t.Any]) -> "Group":
+ ...
+
+ @t.overload
+ def group(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Callable[[t.Callable[..., t.Any]], "Group"]:
+ ...
- def group(self, *args: t.Any, **kwargs: t.Any) ->t.Union[t.Callable[[t.
- Callable[..., t.Any]], 'Group'], 'Group']:
+ def group(
+ self, *args: t.Any, **kwargs: t.Any
+ ) -> t.Union[t.Callable[[t.Callable[..., t.Any]], "Group"], "Group"]:
"""A shortcut decorator for declaring and attaching a group to
the group. This takes the same arguments as :func:`group` and
immediately registers the created group with this group by
@@ -1053,7 +1920,38 @@ class Group(MultiCommand):
.. versionchanged:: 8.0
Added the :attr:`group_class` attribute.
"""
- pass
+ from .decorators import group
+
+ func: t.Optional[t.Callable[..., t.Any]] = None
+
+ if args and callable(args[0]):
+ assert (
+ len(args) == 1 and not kwargs
+ ), "Use 'group(**kwargs)(callable)' to provide arguments."
+ (func,) = args
+ args = ()
+
+ if self.group_class is not None and kwargs.get("cls") is None:
+ if self.group_class is type:
+ kwargs["cls"] = type(self)
+ else:
+ kwargs["cls"] = self.group_class
+
+ def decorator(f: t.Callable[..., t.Any]) -> "Group":
+ cmd: Group = group(*args, **kwargs)(f)
+ self.add_command(cmd)
+ return cmd
+
+ if func is not None:
+ return decorator(func)
+
+ return decorator
+
+ def get_command(self, ctx: Context, cmd_name: str) -> t.Optional[Command]:
+ return self.commands.get(cmd_name)
+
+ def list_commands(self, ctx: Context) -> t.List[str]:
+ return sorted(self.commands)
class CommandCollection(MultiCommand):
@@ -1066,26 +1964,54 @@ class CommandCollection(MultiCommand):
``name`` and ``attrs``.
"""
- def __init__(self, name: t.Optional[str]=None, sources: t.Optional[t.
- List[MultiCommand]]=None, **attrs: t.Any) ->None:
+ def __init__(
+ self,
+ name: t.Optional[str] = None,
+ sources: t.Optional[t.List[MultiCommand]] = None,
+ **attrs: t.Any,
+ ) -> None:
super().__init__(name, **attrs)
+ #: The list of registered multi commands.
self.sources: t.List[MultiCommand] = sources or []
- def add_source(self, multi_cmd: MultiCommand) ->None:
+ def add_source(self, multi_cmd: MultiCommand) -> None:
"""Adds a new multi command to the chain dispatcher."""
- pass
+ self.sources.append(multi_cmd)
+ def get_command(self, ctx: Context, cmd_name: str) -> t.Optional[Command]:
+ for source in self.sources:
+ rv = source.get_command(ctx, cmd_name)
-def _check_iter(value: t.Any) ->t.Iterator[t.Any]:
+ if rv is not None:
+ if self.chain:
+ _check_multicommand(self, cmd_name, rv)
+
+ return rv
+
+ return None
+
+ def list_commands(self, ctx: Context) -> t.List[str]:
+ rv: t.Set[str] = set()
+
+ for source in self.sources:
+ rv.update(source.list_commands(ctx))
+
+ return sorted(rv)
+
+
+def _check_iter(value: t.Any) -> t.Iterator[t.Any]:
"""Check if the value is iterable but not a string. Raises a type
error, or return an iterator over the value.
"""
- pass
+ if isinstance(value, str):
+ raise TypeError
+
+ return iter(value)
class Parameter:
- """A parameter to a command comes in two versions: they are either
- :class:`Option`\\s or :class:`Argument`\\s. Other subclasses are currently
+ r"""A parameter to a command comes in two versions: they are either
+ :class:`Option`\s or :class:`Argument`\s. Other subclasses are currently
not supported by design as some of the internals for parsing are
intentionally not finalized.
@@ -1156,28 +2082,45 @@ class Parameter:
parameter. The old callback format will still work, but it will
raise a warning to give you a chance to migrate the code easier.
"""
- param_type_name = 'parameter'
-
- def __init__(self, param_decls: t.Optional[t.Sequence[str]]=None, type:
- t.Optional[t.Union[types.ParamType, t.Any]]=None, required: bool=
- False, default: t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]=
- None, callback: t.Optional[t.Callable[[Context, 'Parameter', t.Any],
- t.Any]]=None, nargs: t.Optional[int]=None, multiple: bool=False,
- metavar: t.Optional[str]=None, expose_value: bool=True, is_eager:
- bool=False, envvar: t.Optional[t.Union[str, t.Sequence[str]]]=None,
- shell_complete: t.Optional[t.Callable[[Context, 'Parameter', str],
- t.Union[t.List['CompletionItem'], t.List[str]]]]=None) ->None:
+
+ param_type_name = "parameter"
+
+ def __init__(
+ self,
+ param_decls: t.Optional[t.Sequence[str]] = None,
+ type: t.Optional[t.Union[types.ParamType, t.Any]] = None,
+ required: bool = False,
+ default: t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]] = None,
+ callback: t.Optional[t.Callable[[Context, "Parameter", t.Any], t.Any]] = None,
+ nargs: t.Optional[int] = None,
+ multiple: bool = False,
+ metavar: t.Optional[str] = None,
+ expose_value: bool = True,
+ is_eager: bool = False,
+ envvar: t.Optional[t.Union[str, t.Sequence[str]]] = None,
+ shell_complete: t.Optional[
+ t.Callable[
+ [Context, "Parameter", str],
+ t.Union[t.List["CompletionItem"], t.List[str]],
+ ]
+ ] = None,
+ ) -> None:
self.name: t.Optional[str]
self.opts: t.List[str]
self.secondary_opts: t.List[str]
self.name, self.opts, self.secondary_opts = self._parse_decls(
- param_decls or (), expose_value)
+ param_decls or (), expose_value
+ )
self.type: types.ParamType = types.convert_type(type, default)
+
+ # Default nargs to what the type tells us if we have that
+ # information available.
if nargs is None:
if self.type.is_composite:
nargs = self.type.arity
else:
nargs = 1
+
self.required = required
self.callback = callback
self.nargs = nargs
@@ -1188,38 +2131,49 @@ class Parameter:
self.metavar = metavar
self.envvar = envvar
self._custom_shell_complete = shell_complete
+
if __debug__:
if self.type.is_composite and nargs != self.type.arity:
raise ValueError(
- f"'nargs' must be {self.type.arity} (or None) for type {self.type!r}, but it was {nargs}."
- )
+ f"'nargs' must be {self.type.arity} (or None) for"
+ f" type {self.type!r}, but it was {nargs}."
+ )
+
+ # Skip no default or callable default.
check_default = default if not callable(default) else None
+
if check_default is not None:
if multiple:
try:
+ # Only check the first value against nargs.
check_default = next(_check_iter(check_default), None)
except TypeError:
raise ValueError(
"'default' must be a list when 'multiple' is true."
- ) from None
+ ) from None
+
+ # Can be None for multiple with empty default.
if nargs != 1 and check_default is not None:
try:
_check_iter(check_default)
except TypeError:
if multiple:
message = (
- "'default' must be a list of lists when 'multiple' is true and 'nargs' != 1."
- )
+ "'default' must be a list of lists when 'multiple' is"
+ " true and 'nargs' != 1."
+ )
else:
- message = (
- "'default' must be a list when 'nargs' != 1.")
+ message = "'default' must be a list when 'nargs' != 1."
+
raise ValueError(message) from None
+
if nargs > 1 and len(check_default) != nargs:
- subject = 'item length' if multiple else 'length'
+ subject = "item length" if multiple else "length"
raise ValueError(
- f"'default' {subject} must match nargs={nargs}.")
+ f"'default' {subject} must match nargs={nargs}."
+ )
- def to_info_dict(self) ->t.Dict[str, t.Any]:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
"""Gather information that could be useful for a tool generating
user-facing documentation.
@@ -1228,20 +2182,63 @@ class Parameter:
.. versionadded:: 8.0
"""
- pass
-
- def __repr__(self) ->str:
- return f'<{self.__class__.__name__} {self.name}>'
+ return {
+ "name": self.name,
+ "param_type_name": self.param_type_name,
+ "opts": self.opts,
+ "secondary_opts": self.secondary_opts,
+ "type": self.type.to_info_dict(),
+ "required": self.required,
+ "nargs": self.nargs,
+ "multiple": self.multiple,
+ "default": self.default,
+ "envvar": self.envvar,
+ }
+
+ def __repr__(self) -> str:
+ return f"<{self.__class__.__name__} {self.name}>"
+
+ def _parse_decls(
+ self, decls: t.Sequence[str], expose_value: bool
+ ) -> t.Tuple[t.Optional[str], t.List[str], t.List[str]]:
+ raise NotImplementedError()
@property
- def human_readable_name(self) ->str:
+ def human_readable_name(self) -> str:
"""Returns the human readable name of this parameter. This is the
same as the name for options, but the metavar for arguments.
"""
- pass
+ return self.name # type: ignore
+
+ def make_metavar(self) -> str:
+ if self.metavar is not None:
+ return self.metavar
+
+ metavar = self.type.get_metavar(self)
+
+ if metavar is None:
+ metavar = self.type.name.upper()
+
+ if self.nargs != 1:
+ metavar += "..."
+
+ return metavar
- def get_default(self, ctx: Context, call: bool=True) ->t.Optional[t.
- Union[t.Any, t.Callable[[], t.Any]]]:
+ @t.overload
+ def get_default(
+ self, ctx: Context, call: "te.Literal[True]" = True
+ ) -> t.Optional[t.Any]:
+ ...
+
+ @t.overload
+ def get_default(
+ self, ctx: Context, call: bool = ...
+ ) -> t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]:
+ ...
+
+ def get_default(
+ self, ctx: Context, call: bool = True
+ ) -> t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]:
"""Get the default for the parameter. Tries
:meth:`Context.lookup_default` first, then the local default.
@@ -1262,22 +2259,170 @@ class Parameter:
.. versionchanged:: 8.0
Added the ``call`` parameter.
"""
- pass
+ value = ctx.lookup_default(self.name, call=False) # type: ignore
+
+ if value is None:
+ value = self.default
+
+ if call and callable(value):
+ value = value()
- def type_cast_value(self, ctx: Context, value: t.Any) ->t.Any:
+ return value
+
+ def add_to_parser(self, parser: OptionParser, ctx: Context) -> None:
+ raise NotImplementedError()
+
+ def consume_value(
+ self, ctx: Context, opts: t.Mapping[str, t.Any]
+ ) -> t.Tuple[t.Any, ParameterSource]:
+ value = opts.get(self.name) # type: ignore
+ source = ParameterSource.COMMANDLINE
+
+ if value is None:
+ value = self.value_from_envvar(ctx)
+ source = ParameterSource.ENVIRONMENT
+
+ if value is None:
+ value = ctx.lookup_default(self.name) # type: ignore
+ source = ParameterSource.DEFAULT_MAP
+
+ if value is None:
+ value = self.get_default(ctx)
+ source = ParameterSource.DEFAULT
+
+ return value, source
+
+ def type_cast_value(self, ctx: Context, value: t.Any) -> t.Any:
"""Convert and validate a value against the option's
:attr:`type`, :attr:`multiple`, and :attr:`nargs`.
"""
+ if value is None:
+ return () if self.multiple or self.nargs == -1 else None
+
+ def check_iter(value: t.Any) -> t.Iterator[t.Any]:
+ try:
+ return _check_iter(value)
+ except TypeError:
+ # This should only happen when passing in args manually,
+ # the parser should construct an iterable when parsing
+ # the command line.
+ raise BadParameter(
+ _("Value must be an iterable."), ctx=ctx, param=self
+ ) from None
+
+ if self.nargs == 1 or self.type.is_composite:
+
+ def convert(value: t.Any) -> t.Any:
+ return self.type(value, param=self, ctx=ctx)
+
+ elif self.nargs == -1:
+
+ def convert(value: t.Any) -> t.Any: # t.Tuple[t.Any, ...]
+ return tuple(self.type(x, self, ctx) for x in check_iter(value))
+
+ else: # nargs > 1
+
+ def convert(value: t.Any) -> t.Any: # t.Tuple[t.Any, ...]
+ value = tuple(check_iter(value))
+
+ if len(value) != self.nargs:
+ raise BadParameter(
+ ngettext(
+ "Takes {nargs} values but 1 was given.",
+ "Takes {nargs} values but {len} were given.",
+ len(value),
+ ).format(nargs=self.nargs, len=len(value)),
+ ctx=ctx,
+ param=self,
+ )
+
+ return tuple(self.type(x, self, ctx) for x in value)
+
+ if self.multiple:
+ return tuple(convert(x) for x in check_iter(value))
+
+ return convert(value)
+
+ def value_is_missing(self, value: t.Any) -> bool:
+ if value is None:
+ return True
+
+ if (self.nargs != 1 or self.multiple) and value == ():
+ return True
+
+ return False
+
+ def process_value(self, ctx: Context, value: t.Any) -> t.Any:
+ value = self.type_cast_value(ctx, value)
+
+ if self.required and self.value_is_missing(value):
+ raise MissingParameter(ctx=ctx, param=self)
+
+ if self.callback is not None:
+ value = self.callback(ctx, self, value)
+
+ return value
+
+ def resolve_envvar_value(self, ctx: Context) -> t.Optional[str]:
+ if self.envvar is None:
+ return None
+
+ if isinstance(self.envvar, str):
+ rv = os.environ.get(self.envvar)
+
+ if rv:
+ return rv
+ else:
+ for envvar in self.envvar:
+ rv = os.environ.get(envvar)
+
+ if rv:
+ return rv
+
+ return None
+
+ def value_from_envvar(self, ctx: Context) -> t.Optional[t.Any]:
+ rv: t.Optional[t.Any] = self.resolve_envvar_value(ctx)
+
+ if rv is not None and self.nargs != 1:
+ rv = self.type.split_envvar_value(rv)
+
+ return rv
+
+ def handle_parse_result(
+ self, ctx: Context, opts: t.Mapping[str, t.Any], args: t.List[str]
+ ) -> t.Tuple[t.Any, t.List[str]]:
+ with augment_usage_errors(ctx, param=self):
+ value, source = self.consume_value(ctx, opts)
+ ctx.set_parameter_source(self.name, source) # type: ignore
+
+ try:
+ value = self.process_value(ctx, value)
+ except Exception:
+ if not ctx.resilient_parsing:
+ raise
+
+ value = None
+
+ if self.expose_value:
+ ctx.params[self.name] = value # type: ignore
+
+ return value, args
+
+ def get_help_record(self, ctx: Context) -> t.Optional[t.Tuple[str, str]]:
pass
- def get_error_hint(self, ctx: Context) ->str:
+ def get_usage_pieces(self, ctx: Context) -> t.List[str]:
+ return []
+
+ def get_error_hint(self, ctx: Context) -> str:
"""Get a stringified version of the param for use in error messages to
indicate which param caused the error.
"""
- pass
+ hint_list = self.opts or [self.human_readable_name]
+ return " / ".join(f"'{x}'" for x in hint_list)
- def shell_complete(self, ctx: Context, incomplete: str) ->t.List[
- 'CompletionItem']:
+ def shell_complete(self, ctx: Context, incomplete: str) -> t.List["CompletionItem"]:
"""Return a list of completions for the incomplete value. If a
``shell_complete`` function was given during init, it is used.
Otherwise, the :attr:`type`
@@ -1288,7 +2433,17 @@ class Parameter:
.. versionadded:: 8.0
"""
- pass
+ if self._custom_shell_complete is not None:
+ results = self._custom_shell_complete(ctx, self, incomplete)
+
+ if results and isinstance(results[0], str):
+ from click.shell_completion import CompletionItem
+
+ results = [CompletionItem(c) for c in results]
+
+ return t.cast(t.List["CompletionItem"], results)
+
+ return self.type.shell_complete(ctx, self, incomplete)
class Option(Parameter):
@@ -1351,98 +2506,464 @@ class Option(Parameter):
.. versionchanged:: 8.0.1
``type`` is detected from ``flag_value`` if given.
"""
- param_type_name = 'option'
-
- def __init__(self, param_decls: t.Optional[t.Sequence[str]]=None,
- show_default: t.Union[bool, str, None]=None, prompt: t.Union[bool,
- str]=False, confirmation_prompt: t.Union[bool, str]=False,
- prompt_required: bool=True, hide_input: bool=False, is_flag: t.
- Optional[bool]=None, flag_value: t.Optional[t.Any]=None, multiple:
- bool=False, count: bool=False, allow_from_autoenv: bool=True, type:
- t.Optional[t.Union[types.ParamType, t.Any]]=None, help: t.Optional[
- str]=None, hidden: bool=False, show_choices: bool=True, show_envvar:
- bool=False, **attrs: t.Any) ->None:
+
+ param_type_name = "option"
+
+ def __init__(
+ self,
+ param_decls: t.Optional[t.Sequence[str]] = None,
+ show_default: t.Union[bool, str, None] = None,
+ prompt: t.Union[bool, str] = False,
+ confirmation_prompt: t.Union[bool, str] = False,
+ prompt_required: bool = True,
+ hide_input: bool = False,
+ is_flag: t.Optional[bool] = None,
+ flag_value: t.Optional[t.Any] = None,
+ multiple: bool = False,
+ count: bool = False,
+ allow_from_autoenv: bool = True,
+ type: t.Optional[t.Union[types.ParamType, t.Any]] = None,
+ help: t.Optional[str] = None,
+ hidden: bool = False,
+ show_choices: bool = True,
+ show_envvar: bool = False,
+ **attrs: t.Any,
+ ) -> None:
if help:
help = inspect.cleandoc(help)
- default_is_missing = 'default' not in attrs
+
+ default_is_missing = "default" not in attrs
super().__init__(param_decls, type=type, multiple=multiple, **attrs)
+
if prompt is True:
if self.name is None:
raise TypeError("'name' is required with 'prompt=True'.")
- prompt_text: t.Optional[str] = self.name.replace('_', ' '
- ).capitalize()
+
+ prompt_text: t.Optional[str] = self.name.replace("_", " ").capitalize()
elif prompt is False:
prompt_text = None
else:
prompt_text = prompt
+
self.prompt = prompt_text
self.confirmation_prompt = confirmation_prompt
self.prompt_required = prompt_required
self.hide_input = hide_input
self.hidden = hidden
- self._flag_needs_value = (self.prompt is not None and not self.
- prompt_required)
+
+ # If prompt is enabled but not required, then the option can be
+ # used as a flag to indicate using prompt or flag_value.
+ self._flag_needs_value = self.prompt is not None and not self.prompt_required
+
if is_flag is None:
if flag_value is not None:
+ # Implicitly a flag because flag_value was set.
is_flag = True
elif self._flag_needs_value:
+ # Not a flag, but when used as a flag it shows a prompt.
is_flag = False
else:
+ # Implicitly a flag because flag options were given.
is_flag = bool(self.secondary_opts)
elif is_flag is False and not self._flag_needs_value:
+ # Not a flag, and prompt is not enabled, can be used as a
+ # flag if flag_value is set.
self._flag_needs_value = flag_value is not None
+
self.default: t.Union[t.Any, t.Callable[[], t.Any]]
+
if is_flag and default_is_missing and not self.required:
if multiple:
self.default = ()
else:
self.default = False
+
if flag_value is None:
flag_value = not self.default
+
self.type: types.ParamType
if is_flag and type is None:
+ # Re-guess the type from the flag value instead of the
+ # default.
self.type = types.convert_type(None, flag_value)
+
self.is_flag: bool = is_flag
- self.is_bool_flag: bool = is_flag and isinstance(self.type, types.
- BoolParamType)
+ self.is_bool_flag: bool = is_flag and isinstance(self.type, types.BoolParamType)
self.flag_value: t.Any = flag_value
+
+ # Counting
self.count = count
if count:
if type is None:
self.type = types.IntRange(min=0)
if default_is_missing:
self.default = 0
+
self.allow_from_autoenv = allow_from_autoenv
self.help = help
self.show_default = show_default
self.show_choices = show_choices
self.show_envvar = show_envvar
+
if __debug__:
if self.nargs == -1:
- raise TypeError('nargs=-1 is not supported for options.')
+ raise TypeError("nargs=-1 is not supported for options.")
+
if self.prompt and self.is_flag and not self.is_bool_flag:
raise TypeError("'prompt' is not valid for non-boolean flag.")
+
if not self.is_bool_flag and self.secondary_opts:
- raise TypeError(
- 'Secondary flag is not valid for non-boolean flag.')
- if (self.is_bool_flag and self.hide_input and self.prompt is not
- None):
+ raise TypeError("Secondary flag is not valid for non-boolean flag.")
+
+ if self.is_bool_flag and self.hide_input and self.prompt is not None:
raise TypeError(
"'prompt' with 'hide_input' is not valid for boolean flag."
- )
+ )
+
if self.count:
if self.multiple:
raise TypeError("'count' is not valid with 'multiple'.")
+
if self.is_flag:
raise TypeError("'count' is not valid with 'is_flag'.")
- def prompt_for_value(self, ctx: Context) ->t.Any:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict.update(
+ help=self.help,
+ prompt=self.prompt,
+ is_flag=self.is_flag,
+ flag_value=self.flag_value,
+ count=self.count,
+ hidden=self.hidden,
+ )
+ return info_dict
+
+ def _parse_decls(
+ self, decls: t.Sequence[str], expose_value: bool
+ ) -> t.Tuple[t.Optional[str], t.List[str], t.List[str]]:
+ opts = []
+ secondary_opts = []
+ name = None
+ possible_names = []
+
+ for decl in decls:
+ if decl.isidentifier():
+ if name is not None:
+ raise TypeError(f"Name '{name}' defined twice")
+ name = decl
+ else:
+ split_char = ";" if decl[:1] == "/" else "/"
+ if split_char in decl:
+ first, second = decl.split(split_char, 1)
+ first = first.rstrip()
+ if first:
+ possible_names.append(split_opt(first))
+ opts.append(first)
+ second = second.lstrip()
+ if second:
+ secondary_opts.append(second.lstrip())
+ if first == second:
+ raise ValueError(
+ f"Boolean option {decl!r} cannot use the"
+ " same flag for true/false."
+ )
+ else:
+ possible_names.append(split_opt(decl))
+ opts.append(decl)
+
+ if name is None and possible_names:
+ possible_names.sort(key=lambda x: -len(x[0])) # group long options first
+ name = possible_names[0][1].replace("-", "_").lower()
+ if not name.isidentifier():
+ name = None
+
+ if name is None:
+ if not expose_value:
+ return None, opts, secondary_opts
+ raise TypeError("Could not determine name for option")
+
+ if not opts and not secondary_opts:
+ raise TypeError(
+ f"No options defined but a name was passed ({name})."
+ " Did you mean to declare an argument instead? Did"
+ f" you mean to pass '--{name}'?"
+ )
+
+ return name, opts, secondary_opts
+
+ def add_to_parser(self, parser: OptionParser, ctx: Context) -> None:
+ if self.multiple:
+ action = "append"
+ elif self.count:
+ action = "count"
+ else:
+ action = "store"
+
+ if self.is_flag:
+ action = f"{action}_const"
+
+ if self.is_bool_flag and self.secondary_opts:
+ parser.add_option(
+ obj=self, opts=self.opts, dest=self.name, action=action, const=True
+ )
+ parser.add_option(
+ obj=self,
+ opts=self.secondary_opts,
+ dest=self.name,
+ action=action,
+ const=False,
+ )
+ else:
+ parser.add_option(
+ obj=self,
+ opts=self.opts,
+ dest=self.name,
+ action=action,
+ const=self.flag_value,
+ )
+ else:
+ parser.add_option(
+ obj=self,
+ opts=self.opts,
+ dest=self.name,
+ action=action,
+ nargs=self.nargs,
+ )
+
+ def get_help_record(self, ctx: Context) -> t.Optional[t.Tuple[str, str]]:
+ if self.hidden:
+ return None
+
+ any_prefix_is_slash = False
+
+ def _write_opts(opts: t.Sequence[str]) -> str:
+ nonlocal any_prefix_is_slash
+
+ rv, any_slashes = join_options(opts)
+
+ if any_slashes:
+ any_prefix_is_slash = True
+
+ if not self.is_flag and not self.count:
+ rv += f" {self.make_metavar()}"
+
+ return rv
+
+ rv = [_write_opts(self.opts)]
+
+ if self.secondary_opts:
+ rv.append(_write_opts(self.secondary_opts))
+
+ help = self.help or ""
+ extra = []
+
+ if self.show_envvar:
+ envvar = self.envvar
+
+ if envvar is None:
+ if (
+ self.allow_from_autoenv
+ and ctx.auto_envvar_prefix is not None
+ and self.name is not None
+ ):
+ envvar = f"{ctx.auto_envvar_prefix}_{self.name.upper()}"
+
+ if envvar is not None:
+ var_str = (
+ envvar
+ if isinstance(envvar, str)
+ else ", ".join(str(d) for d in envvar)
+ )
+ extra.append(_("env var: {var}").format(var=var_str))
+
+ # Temporarily enable resilient parsing to avoid type casting
+ # failing for the default. Might be possible to extend this to
+ # help formatting in general.
+ resilient = ctx.resilient_parsing
+ ctx.resilient_parsing = True
+
+ try:
+ default_value = self.get_default(ctx, call=False)
+ finally:
+ ctx.resilient_parsing = resilient
+
+ show_default = False
+ show_default_is_str = False
+
+ if self.show_default is not None:
+ if isinstance(self.show_default, str):
+ show_default_is_str = show_default = True
+ else:
+ show_default = self.show_default
+ elif ctx.show_default is not None:
+ show_default = ctx.show_default
+
+ if show_default_is_str or (show_default and (default_value is not None)):
+ if show_default_is_str:
+ default_string = f"({self.show_default})"
+ elif isinstance(default_value, (list, tuple)):
+ default_string = ", ".join(str(d) for d in default_value)
+ elif inspect.isfunction(default_value):
+ default_string = _("(dynamic)")
+ elif self.is_bool_flag and self.secondary_opts:
+ # For boolean flags that have distinct True/False opts,
+ # use the opt without prefix instead of the value.
+ default_string = split_opt(
+ (self.opts if self.default else self.secondary_opts)[0]
+ )[1]
+ elif self.is_bool_flag and not self.secondary_opts and not default_value:
+ default_string = ""
+ else:
+ default_string = str(default_value)
+
+ if default_string:
+ extra.append(_("default: {default}").format(default=default_string))
+
+ if (
+ isinstance(self.type, types._NumberRangeBase)
+ # skip count with default range type
+ and not (self.count and self.type.min == 0 and self.type.max is None)
+ ):
+ range_str = self.type._describe_range()
+
+ if range_str:
+ extra.append(range_str)
+
+ if self.required:
+ extra.append(_("required"))
+
+ if extra:
+ extra_str = "; ".join(extra)
+ help = f"{help} [{extra_str}]" if help else f"[{extra_str}]"
+
+ return ("; " if any_prefix_is_slash else " / ").join(rv), help
+
+ @t.overload
+ def get_default(
+ self, ctx: Context, call: "te.Literal[True]" = True
+ ) -> t.Optional[t.Any]:
+ ...
+
+ @t.overload
+ def get_default(
+ self, ctx: Context, call: bool = ...
+ ) -> t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]:
+ ...
+
+ def get_default(
+ self, ctx: Context, call: bool = True
+ ) -> t.Optional[t.Union[t.Any, t.Callable[[], t.Any]]]:
+ # If we're a non boolean flag our default is more complex because
+ # we need to look at all flags in the same group to figure out
+ # if we're the default one in which case we return the flag
+ # value as default.
+ if self.is_flag and not self.is_bool_flag:
+ for param in ctx.command.params:
+ if param.name == self.name and param.default:
+ return t.cast(Option, param).flag_value
+
+ return None
+
+ return super().get_default(ctx, call=call)
+
+ def prompt_for_value(self, ctx: Context) -> t.Any:
"""This is an alternative flow that can be activated in the full
value processing if a value does not exist. It will prompt the
user until a valid value exists and then returns the processed
value as result.
"""
- pass
+ assert self.prompt is not None
+
+ # Calculate the default before prompting anything to be stable.
+ default = self.get_default(ctx)
+
+ # If this is a prompt for a flag we need to handle this
+ # differently.
+ if self.is_bool_flag:
+ return confirm(self.prompt, default)
+
+ return prompt(
+ self.prompt,
+ default=default,
+ type=self.type,
+ hide_input=self.hide_input,
+ show_choices=self.show_choices,
+ confirmation_prompt=self.confirmation_prompt,
+ value_proc=lambda x: self.process_value(ctx, x),
+ )
+
+ def resolve_envvar_value(self, ctx: Context) -> t.Optional[str]:
+ rv = super().resolve_envvar_value(ctx)
+
+ if rv is not None:
+ return rv
+
+ if (
+ self.allow_from_autoenv
+ and ctx.auto_envvar_prefix is not None
+ and self.name is not None
+ ):
+ envvar = f"{ctx.auto_envvar_prefix}_{self.name.upper()}"
+ rv = os.environ.get(envvar)
+
+ if rv:
+ return rv
+
+ return None
+
+ def value_from_envvar(self, ctx: Context) -> t.Optional[t.Any]:
+ rv: t.Optional[t.Any] = self.resolve_envvar_value(ctx)
+
+ if rv is None:
+ return None
+
+ value_depth = (self.nargs != 1) + bool(self.multiple)
+
+ if value_depth > 0:
+ rv = self.type.split_envvar_value(rv)
+
+ if self.multiple and self.nargs != 1:
+ rv = batch(rv, self.nargs)
+
+ return rv
+
+ def consume_value(
+ self, ctx: Context, opts: t.Mapping[str, "Parameter"]
+ ) -> t.Tuple[t.Any, ParameterSource]:
+ value, source = super().consume_value(ctx, opts)
+
+ # The parser will emit a sentinel value if the option can be
+ # given as a flag without a value. This is different from None
+ # to distinguish from the flag not being given at all.
+ if value is _flag_needs_value:
+ if self.prompt is not None and not ctx.resilient_parsing:
+ value = self.prompt_for_value(ctx)
+ source = ParameterSource.PROMPT
+ else:
+ value = self.flag_value
+ source = ParameterSource.COMMANDLINE
+
+ elif (
+ self.multiple
+ and value is not None
+ and any(v is _flag_needs_value for v in value)
+ ):
+ value = [self.flag_value if v is _flag_needs_value else v for v in value]
+ source = ParameterSource.COMMANDLINE
+
+ # The value wasn't set, or used the param's default, prompt if
+ # prompting is enabled.
+ elif (
+ source in {None, ParameterSource.DEFAULT}
+ and self.prompt is not None
+ and (self.required or self.prompt_required)
+ and not ctx.resilient_parsing
+ ):
+ value = self.prompt_for_value(ctx)
+ source = ParameterSource.PROMPT
+
+ return value, source
class Argument(Parameter):
@@ -1452,19 +2973,70 @@ class Argument(Parameter):
All parameters are passed onwards to the constructor of :class:`Parameter`.
"""
- param_type_name = 'argument'
- def __init__(self, param_decls: t.Sequence[str], required: t.Optional[
- bool]=None, **attrs: t.Any) ->None:
+ param_type_name = "argument"
+
+ def __init__(
+ self,
+ param_decls: t.Sequence[str],
+ required: t.Optional[bool] = None,
+ **attrs: t.Any,
+ ) -> None:
if required is None:
- if attrs.get('default') is not None:
+ if attrs.get("default") is not None:
required = False
else:
- required = attrs.get('nargs', 1) > 0
- if 'multiple' in attrs:
- raise TypeError(
- "__init__() got an unexpected keyword argument 'multiple'.")
+ required = attrs.get("nargs", 1) > 0
+
+ if "multiple" in attrs:
+ raise TypeError("__init__() got an unexpected keyword argument 'multiple'.")
+
super().__init__(param_decls, required=required, **attrs)
+
if __debug__:
if self.default is not None and self.nargs == -1:
raise TypeError("'default' is not supported for nargs=-1.")
+
+ @property
+ def human_readable_name(self) -> str:
+ if self.metavar is not None:
+ return self.metavar
+ return self.name.upper() # type: ignore
+
+ def make_metavar(self) -> str:
+ if self.metavar is not None:
+ return self.metavar
+ var = self.type.get_metavar(self)
+ if not var:
+ var = self.name.upper() # type: ignore
+ if not self.required:
+ var = f"[{var}]"
+ if self.nargs != 1:
+ var += "..."
+ return var
+
+ def _parse_decls(
+ self, decls: t.Sequence[str], expose_value: bool
+ ) -> t.Tuple[t.Optional[str], t.List[str], t.List[str]]:
+ if not decls:
+ if not expose_value:
+ return None, [], []
+ raise TypeError("Could not determine name for argument")
+ if len(decls) == 1:
+ name = arg = decls[0]
+ name = name.replace("-", "_").lower()
+ else:
+ raise TypeError(
+ "Arguments take exactly one parameter declaration, got"
+ f" {len(decls)}."
+ )
+ return name, [arg], []
+
+ def get_usage_pieces(self, ctx: Context) -> t.List[str]:
+ return [self.make_metavar()]
+
+ def get_error_hint(self, ctx: Context) -> str:
+ return f"'{self.make_metavar()}'"
+
+ def add_to_parser(self, parser: OptionParser, ctx: Context) -> None:
+ parser.add_argument(dest=self.name, nargs=self.nargs, obj=self)
diff --git a/src/click/decorators.py b/src/click/decorators.py
index ec1bfc9..d9bba95 100644
--- a/src/click/decorators.py
+++ b/src/click/decorators.py
@@ -3,6 +3,7 @@ import types
import typing as t
from functools import update_wrapper
from gettext import gettext as _
+
from .core import Argument
from .core import Command
from .core import Context
@@ -11,35 +12,44 @@ from .core import Option
from .core import Parameter
from .globals import get_current_context
from .utils import echo
+
if t.TYPE_CHECKING:
import typing_extensions as te
- P = te.ParamSpec('P')
-R = t.TypeVar('R')
-T = t.TypeVar('T')
+
+ P = te.ParamSpec("P")
+
+R = t.TypeVar("R")
+T = t.TypeVar("T")
_AnyCallable = t.Callable[..., t.Any]
-FC = t.TypeVar('FC', bound=t.Union[_AnyCallable, Command])
+FC = t.TypeVar("FC", bound=t.Union[_AnyCallable, Command])
-def pass_context(f: 't.Callable[te.Concatenate[Context, P], R]'
- ) ->'t.Callable[P, R]':
+def pass_context(f: "t.Callable[te.Concatenate[Context, P], R]") -> "t.Callable[P, R]":
"""Marks a callback as wanting to receive the current context
object as first argument.
"""
- pass
+ def new_func(*args: "P.args", **kwargs: "P.kwargs") -> "R":
+ return f(get_current_context(), *args, **kwargs)
+
+ return update_wrapper(new_func, f)
-def pass_obj(f: 't.Callable[te.Concatenate[t.Any, P], R]'
- ) ->'t.Callable[P, R]':
+
+def pass_obj(f: "t.Callable[te.Concatenate[t.Any, P], R]") -> "t.Callable[P, R]":
"""Similar to :func:`pass_context`, but only pass the object on the
context onwards (:attr:`Context.obj`). This is useful if that object
represents the state of a nested system.
"""
- pass
+ def new_func(*args: "P.args", **kwargs: "P.kwargs") -> "R":
+ return f(get_current_context().obj, *args, **kwargs)
+
+ return update_wrapper(new_func, f)
-def make_pass_decorator(object_type: t.Type[T], ensure: bool=False
- ) ->t.Callable[['t.Callable[te.Concatenate[T, P], R]'], 't.Callable[P, R]'
- ]:
+
+def make_pass_decorator(
+ object_type: t.Type[T], ensure: bool = False
+) -> t.Callable[["t.Callable[te.Concatenate[T, P], R]"], "t.Callable[P, R]"]:
"""Given an object type this creates a decorator that will work
similar to :func:`pass_obj` but instead of passing the object of the
current context, it will find the innermost context of type
@@ -61,11 +71,34 @@ def make_pass_decorator(object_type: t.Type[T], ensure: bool=False
:param ensure: if set to `True`, a new object will be created and
remembered on the context if it's not there yet.
"""
- pass
+ def decorator(f: "t.Callable[te.Concatenate[T, P], R]") -> "t.Callable[P, R]":
+ def new_func(*args: "P.args", **kwargs: "P.kwargs") -> "R":
+ ctx = get_current_context()
-def pass_meta_key(key: str, *, doc_description: t.Optional[str]=None
- ) ->'t.Callable[[t.Callable[te.Concatenate[t.Any, P], R]], t.Callable[P, R]]':
+ obj: t.Optional[T]
+ if ensure:
+ obj = ctx.ensure_object(object_type)
+ else:
+ obj = ctx.find_object(object_type)
+
+ if obj is None:
+ raise RuntimeError(
+ "Managed to invoke callback without a context"
+ f" object of type {object_type.__name__!r}"
+ " existing."
+ )
+
+ return ctx.invoke(f, obj, *args, **kwargs)
+
+ return update_wrapper(new_func, f)
+
+ return decorator # type: ignore[return-value]
+
+
+def pass_meta_key(
+ key: str, *, doc_description: t.Optional[str] = None
+) -> "t.Callable[[t.Callable[te.Concatenate[t.Any, P], R]], t.Callable[P, R]]":
"""Create a decorator that passes a key from
:attr:`click.Context.meta` as the first argument to the decorated
function.
@@ -77,18 +110,72 @@ def pass_meta_key(key: str, *, doc_description: t.Optional[str]=None
.. versionadded:: 8.0
"""
- pass
+ def decorator(f: "t.Callable[te.Concatenate[t.Any, P], R]") -> "t.Callable[P, R]":
+ def new_func(*args: "P.args", **kwargs: "P.kwargs") -> R:
+ ctx = get_current_context()
+ obj = ctx.meta[key]
+ return ctx.invoke(f, obj, *args, **kwargs)
+
+ return update_wrapper(new_func, f)
+
+ if doc_description is None:
+ doc_description = f"the {key!r} key from :attr:`click.Context.meta`"
-CmdType = t.TypeVar('CmdType', bound=Command)
+ decorator.__doc__ = (
+ f"Decorator that passes {doc_description} as the first argument"
+ " to the decorated function."
+ )
+ return decorator # type: ignore[return-value]
-def command(name: t.Union[t.Optional[str], _AnyCallable]=None, cls: t.
- Optional[t.Type[CmdType]]=None, **attrs: t.Any) ->t.Union[Command, t.
- Callable[[_AnyCallable], t.Union[Command, CmdType]]]:
- """Creates a new :class:`Command` and uses the decorated function as
+CmdType = t.TypeVar("CmdType", bound=Command)
+
+
+# variant: no call, directly as decorator for a function.
+@t.overload
+def command(name: _AnyCallable) -> Command:
+ ...
+
+
+# variant: with positional name and with positional or keyword cls argument:
+# @command(namearg, CommandCls, ...) or @command(namearg, cls=CommandCls, ...)
+@t.overload
+def command(
+ name: t.Optional[str],
+ cls: t.Type[CmdType],
+ **attrs: t.Any,
+) -> t.Callable[[_AnyCallable], CmdType]:
+ ...
+
+
+# variant: name omitted, cls _must_ be a keyword argument, @command(cls=CommandCls, ...)
+@t.overload
+def command(
+ name: None = None,
+ *,
+ cls: t.Type[CmdType],
+ **attrs: t.Any,
+) -> t.Callable[[_AnyCallable], CmdType]:
+ ...
+
+
+# variant: with optional string name, no cls argument provided.
+@t.overload
+def command(
+ name: t.Optional[str] = ..., cls: None = None, **attrs: t.Any
+) -> t.Callable[[_AnyCallable], Command]:
+ ...
+
+
+def command(
+ name: t.Union[t.Optional[str], _AnyCallable] = None,
+ cls: t.Optional[t.Type[CmdType]] = None,
+ **attrs: t.Any,
+) -> t.Union[Command, t.Callable[[_AnyCallable], t.Union[Command, CmdType]]]:
+ r"""Creates a new :class:`Command` and uses the decorated function as
callback. This will also automatically attach all decorated
- :func:`option`\\s and :func:`argument`\\s as parameters to the command.
+ :func:`option`\s and :func:`argument`\s as parameters to the command.
The name of the command defaults to the name of the function with
underscores replaced by dashes. If you want to change that, you can
@@ -114,15 +201,99 @@ def command(name: t.Union[t.Optional[str], _AnyCallable]=None, cls: t.
The ``params`` argument can be used. Decorated params are
appended to the end of the list.
"""
- pass
+ func: t.Optional[t.Callable[[_AnyCallable], t.Any]] = None
+
+ if callable(name):
+ func = name
+ name = None
+ assert cls is None, "Use 'command(cls=cls)(callable)' to specify a class."
+ assert not attrs, "Use 'command(**kwargs)(callable)' to provide arguments."
+
+ if cls is None:
+ cls = t.cast(t.Type[CmdType], Command)
+
+ def decorator(f: _AnyCallable) -> CmdType:
+ if isinstance(f, Command):
+ raise TypeError("Attempted to convert a callback into a command twice.")
+
+ attr_params = attrs.pop("params", None)
+ params = attr_params if attr_params is not None else []
+
+ try:
+ decorator_params = f.__click_params__ # type: ignore
+ except AttributeError:
+ pass
+ else:
+ del f.__click_params__ # type: ignore
+ params.extend(reversed(decorator_params))
+
+ if attrs.get("help") is None:
+ attrs["help"] = f.__doc__
-GrpType = t.TypeVar('GrpType', bound=Group)
+ if t.TYPE_CHECKING:
+ assert cls is not None
+ assert not callable(name)
+ cmd = cls(
+ name=name or f.__name__.lower().replace("_", "-"),
+ callback=f,
+ params=params,
+ **attrs,
+ )
+ cmd.__doc__ = f.__doc__
+ return cmd
-def group(name: t.Union[str, _AnyCallable, None]=None, cls: t.Optional[t.
- Type[GrpType]]=None, **attrs: t.Any) ->t.Union[Group, t.Callable[[
- _AnyCallable], t.Union[Group, GrpType]]]:
+ if func is not None:
+ return decorator(func)
+
+ return decorator
+
+
+GrpType = t.TypeVar("GrpType", bound=Group)
+
+
+# variant: no call, directly as decorator for a function.
+@t.overload
+def group(name: _AnyCallable) -> Group:
+ ...
+
+
+# variant: with positional name and with positional or keyword cls argument:
+# @group(namearg, GroupCls, ...) or @group(namearg, cls=GroupCls, ...)
+@t.overload
+def group(
+ name: t.Optional[str],
+ cls: t.Type[GrpType],
+ **attrs: t.Any,
+) -> t.Callable[[_AnyCallable], GrpType]:
+ ...
+
+
+# variant: name omitted, cls _must_ be a keyword argument, @group(cmd=GroupCls, ...)
+@t.overload
+def group(
+ name: None = None,
+ *,
+ cls: t.Type[GrpType],
+ **attrs: t.Any,
+) -> t.Callable[[_AnyCallable], GrpType]:
+ ...
+
+
+# variant: with optional string name, no cls argument provided.
+@t.overload
+def group(
+ name: t.Optional[str] = ..., cls: None = None, **attrs: t.Any
+) -> t.Callable[[_AnyCallable], Group]:
+ ...
+
+
+def group(
+ name: t.Union[str, _AnyCallable, None] = None,
+ cls: t.Optional[t.Type[GrpType]] = None,
+ **attrs: t.Any,
+) -> t.Union[Group, t.Callable[[_AnyCallable], t.Union[Group, GrpType]]]:
"""Creates a new :class:`Group` with a function as callback. This
works otherwise the same as :func:`command` just that the `cls`
parameter is set to :class:`Group`.
@@ -130,11 +301,28 @@ def group(name: t.Union[str, _AnyCallable, None]=None, cls: t.Optional[t.
.. versionchanged:: 8.1
This decorator can be applied without parentheses.
"""
- pass
+ if cls is None:
+ cls = t.cast(t.Type[GrpType], Group)
+
+ if callable(name):
+ return command(cls=cls, **attrs)(name)
+
+ return command(name, cls, **attrs)
+
+
+def _param_memo(f: t.Callable[..., t.Any], param: Parameter) -> None:
+ if isinstance(f, Command):
+ f.params.append(param)
+ else:
+ if not hasattr(f, "__click_params__"):
+ f.__click_params__ = [] # type: ignore
+ f.__click_params__.append(param) # type: ignore
-def argument(*param_decls: str, cls: t.Optional[t.Type[Argument]]=None, **
- attrs: t.Any) ->t.Callable[[FC], FC]:
+
+def argument(
+ *param_decls: str, cls: t.Optional[t.Type[Argument]] = None, **attrs: t.Any
+) -> t.Callable[[FC], FC]:
"""Attaches an argument to the command. All positional arguments are
passed as parameter declarations to :class:`Argument`; all keyword
arguments are forwarded unchanged (except ``cls``).
@@ -150,11 +338,19 @@ def argument(*param_decls: str, cls: t.Optional[t.Type[Argument]]=None, **
``cls``.
:param attrs: Passed as keyword arguments to the constructor of ``cls``.
"""
- pass
+ if cls is None:
+ cls = Argument
+
+ def decorator(f: FC) -> FC:
+ _param_memo(f, cls(param_decls, **attrs))
+ return f
+ return decorator
-def option(*param_decls: str, cls: t.Optional[t.Type[Option]]=None, **attrs:
- t.Any) ->t.Callable[[FC], FC]:
+
+def option(
+ *param_decls: str, cls: t.Optional[t.Type[Option]] = None, **attrs: t.Any
+) -> t.Callable[[FC], FC]:
"""Attaches an option to the command. All positional arguments are
passed as parameter declarations to :class:`Option`; all keyword
arguments are forwarded unchanged (except ``cls``).
@@ -170,11 +366,17 @@ def option(*param_decls: str, cls: t.Optional[t.Type[Option]]=None, **attrs:
``cls``.
:param attrs: Passed as keyword arguments to the constructor of ``cls``.
"""
- pass
+ if cls is None:
+ cls = Option
+
+ def decorator(f: FC) -> FC:
+ _param_memo(f, cls(param_decls, **attrs))
+ return f
+ return decorator
-def confirmation_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[
- FC], FC]:
+
+def confirmation_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]:
"""Add a ``--yes`` option which shows a prompt before continuing if
not passed. If the prompt is declined, the program will exit.
@@ -182,10 +384,23 @@ def confirmation_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[
value ``"--yes"``.
:param kwargs: Extra arguments are passed to :func:`option`.
"""
- pass
+
+ def callback(ctx: Context, param: Parameter, value: bool) -> None:
+ if not value:
+ ctx.abort()
+
+ if not param_decls:
+ param_decls = ("--yes",)
+
+ kwargs.setdefault("is_flag", True)
+ kwargs.setdefault("callback", callback)
+ kwargs.setdefault("expose_value", False)
+ kwargs.setdefault("prompt", "Do you want to continue?")
+ kwargs.setdefault("help", "Confirm the action without prompting.")
+ return option(*param_decls, **kwargs)
-def password_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[FC], FC]:
+def password_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]:
"""Add a ``--password`` option which prompts for a password, hiding
input and asking to enter the value again for confirmation.
@@ -193,12 +408,23 @@ def password_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[FC], FC]:
value ``"--password"``.
:param kwargs: Extra arguments are passed to :func:`option`.
"""
- pass
-
-
-def version_option(version: t.Optional[str]=None, *param_decls: str,
- package_name: t.Optional[str]=None, prog_name: t.Optional[str]=None,
- message: t.Optional[str]=None, **kwargs: t.Any) ->t.Callable[[FC], FC]:
+ if not param_decls:
+ param_decls = ("--password",)
+
+ kwargs.setdefault("prompt", True)
+ kwargs.setdefault("confirmation_prompt", True)
+ kwargs.setdefault("hide_input", True)
+ return option(*param_decls, **kwargs)
+
+
+def version_option(
+ version: t.Optional[str] = None,
+ *param_decls: str,
+ package_name: t.Optional[str] = None,
+ prog_name: t.Optional[str] = None,
+ message: t.Optional[str] = None,
+ **kwargs: t.Any,
+) -> t.Callable[[FC], FC]:
"""Add a ``--version`` option which immediately prints the version
number and exits the program.
@@ -235,10 +461,76 @@ def version_option(version: t.Optional[str]=None, *param_decls: str,
point name. The Python package name must match the installed
package name, or be passed with ``package_name=``.
"""
- pass
-
-
-def help_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[FC], FC]:
+ if message is None:
+ message = _("%(prog)s, version %(version)s")
+
+ if version is None and package_name is None:
+ frame = inspect.currentframe()
+ f_back = frame.f_back if frame is not None else None
+ f_globals = f_back.f_globals if f_back is not None else None
+ # break reference cycle
+ # https://docs.python.org/3/library/inspect.html#the-interpreter-stack
+ del frame
+
+ if f_globals is not None:
+ package_name = f_globals.get("__name__")
+
+ if package_name == "__main__":
+ package_name = f_globals.get("__package__")
+
+ if package_name:
+ package_name = package_name.partition(".")[0]
+
+ def callback(ctx: Context, param: Parameter, value: bool) -> None:
+ if not value or ctx.resilient_parsing:
+ return
+
+ nonlocal prog_name
+ nonlocal version
+
+ if prog_name is None:
+ prog_name = ctx.find_root().info_name
+
+ if version is None and package_name is not None:
+ metadata: t.Optional[types.ModuleType]
+
+ try:
+ from importlib import metadata # type: ignore
+ except ImportError:
+ # Python < 3.8
+ import importlib_metadata as metadata # type: ignore
+
+ try:
+ version = metadata.version(package_name) # type: ignore
+ except metadata.PackageNotFoundError: # type: ignore
+ raise RuntimeError(
+ f"{package_name!r} is not installed. Try passing"
+ " 'package_name' instead."
+ ) from None
+
+ if version is None:
+ raise RuntimeError(
+ f"Could not determine the version for {package_name!r} automatically."
+ )
+
+ echo(
+ message % {"prog": prog_name, "package": package_name, "version": version},
+ color=ctx.color,
+ )
+ ctx.exit()
+
+ if not param_decls:
+ param_decls = ("--version",)
+
+ kwargs.setdefault("is_flag", True)
+ kwargs.setdefault("expose_value", False)
+ kwargs.setdefault("is_eager", True)
+ kwargs.setdefault("help", _("Show the version and exit."))
+ kwargs["callback"] = callback
+ return option(*param_decls, **kwargs)
+
+
+def help_option(*param_decls: str, **kwargs: t.Any) -> t.Callable[[FC], FC]:
"""Add a ``--help`` option which immediately prints the help page
and exits the program.
@@ -250,4 +542,20 @@ def help_option(*param_decls: str, **kwargs: t.Any) ->t.Callable[[FC], FC]:
value ``"--help"``.
:param kwargs: Extra arguments are passed to :func:`option`.
"""
- pass
+
+ def callback(ctx: Context, param: Parameter, value: bool) -> None:
+ if not value or ctx.resilient_parsing:
+ return
+
+ echo(ctx.get_help(), color=ctx.color)
+ ctx.exit()
+
+ if not param_decls:
+ param_decls = ("--help",)
+
+ kwargs.setdefault("is_flag", True)
+ kwargs.setdefault("expose_value", False)
+ kwargs.setdefault("is_eager", True)
+ kwargs.setdefault("help", _("Show this message and exit."))
+ kwargs["callback"] = callback
+ return option(*param_decls, **kwargs)
diff --git a/src/click/exceptions.py b/src/click/exceptions.py
index 3ce2a80..fe68a36 100644
--- a/src/click/exceptions.py
+++ b/src/click/exceptions.py
@@ -1,26 +1,48 @@
import typing as t
from gettext import gettext as _
from gettext import ngettext
+
from ._compat import get_text_stderr
from .utils import echo
from .utils import format_filename
+
if t.TYPE_CHECKING:
from .core import Command
from .core import Context
from .core import Parameter
+def _join_param_hints(
+ param_hint: t.Optional[t.Union[t.Sequence[str], str]]
+) -> t.Optional[str]:
+ if param_hint is not None and not isinstance(param_hint, str):
+ return " / ".join(repr(x) for x in param_hint)
+
+ return param_hint
+
+
class ClickException(Exception):
"""An exception that Click can handle and show to the user."""
+
+ #: The exit code for this exception.
exit_code = 1
- def __init__(self, message: str) ->None:
+ def __init__(self, message: str) -> None:
super().__init__(message)
self.message = message
- def __str__(self) ->str:
+ def format_message(self) -> str:
return self.message
+ def __str__(self) -> str:
+ return self.message
+
+ def show(self, file: t.Optional[t.IO[t.Any]] = None) -> None:
+ if file is None:
+ file = get_text_stderr()
+
+ echo(_("Error: {message}").format(message=self.format_message()), file=file)
+
class UsageError(ClickException):
"""An internal exception that signals a usage error. This typically
@@ -30,13 +52,35 @@ class UsageError(ClickException):
:param ctx: optionally the context that caused this error. Click will
fill in the context automatically in some situations.
"""
+
exit_code = 2
- def __init__(self, message: str, ctx: t.Optional['Context']=None) ->None:
+ def __init__(self, message: str, ctx: t.Optional["Context"] = None) -> None:
super().__init__(message)
self.ctx = ctx
- self.cmd: t.Optional['Command'
- ] = self.ctx.command if self.ctx else None
+ self.cmd: t.Optional["Command"] = self.ctx.command if self.ctx else None
+
+ def show(self, file: t.Optional[t.IO[t.Any]] = None) -> None:
+ if file is None:
+ file = get_text_stderr()
+ color = None
+ hint = ""
+ if (
+ self.ctx is not None
+ and self.ctx.command.get_help_option(self.ctx) is not None
+ ):
+ hint = _("Try '{command} {option}' for help.").format(
+ command=self.ctx.command_path, option=self.ctx.help_option_names[0]
+ )
+ hint = f"{hint}\n"
+ if self.ctx is not None:
+ color = self.ctx.color
+ echo(f"{self.ctx.get_usage()}\n{hint}", file=file, color=color)
+ echo(
+ _("Error: {message}").format(message=self.format_message()),
+ file=file,
+ color=color,
+ )
class BadParameter(UsageError):
@@ -57,12 +101,29 @@ class BadParameter(UsageError):
each item is quoted and separated.
"""
- def __init__(self, message: str, ctx: t.Optional['Context']=None, param:
- t.Optional['Parameter']=None, param_hint: t.Optional[str]=None) ->None:
+ def __init__(
+ self,
+ message: str,
+ ctx: t.Optional["Context"] = None,
+ param: t.Optional["Parameter"] = None,
+ param_hint: t.Optional[str] = None,
+ ) -> None:
super().__init__(message, ctx)
self.param = param
self.param_hint = param_hint
+ def format_message(self) -> str:
+ if self.param_hint is not None:
+ param_hint = self.param_hint
+ elif self.param is not None:
+ param_hint = self.param.get_error_hint(self.ctx) # type: ignore
+ else:
+ return _("Invalid value: {message}").format(message=self.message)
+
+ return _("Invalid value for {param_hint}: {message}").format(
+ param_hint=_join_param_hints(param_hint), message=self.message
+ )
+
class MissingParameter(BadParameter):
"""Raised if click required an option or argument but it was not
@@ -76,17 +137,59 @@ class MissingParameter(BadParameter):
``'option'`` or ``'argument'``.
"""
- def __init__(self, message: t.Optional[str]=None, ctx: t.Optional[
- 'Context']=None, param: t.Optional['Parameter']=None, param_hint: t
- .Optional[str]=None, param_type: t.Optional[str]=None) ->None:
- super().__init__(message or '', ctx, param, param_hint)
+ def __init__(
+ self,
+ message: t.Optional[str] = None,
+ ctx: t.Optional["Context"] = None,
+ param: t.Optional["Parameter"] = None,
+ param_hint: t.Optional[str] = None,
+ param_type: t.Optional[str] = None,
+ ) -> None:
+ super().__init__(message or "", ctx, param, param_hint)
self.param_type = param_type
- def __str__(self) ->str:
+ def format_message(self) -> str:
+ if self.param_hint is not None:
+ param_hint: t.Optional[str] = self.param_hint
+ elif self.param is not None:
+ param_hint = self.param.get_error_hint(self.ctx) # type: ignore
+ else:
+ param_hint = None
+
+ param_hint = _join_param_hints(param_hint)
+ param_hint = f" {param_hint}" if param_hint else ""
+
+ param_type = self.param_type
+ if param_type is None and self.param is not None:
+ param_type = self.param.param_type_name
+
+ msg = self.message
+ if self.param is not None:
+ msg_extra = self.param.type.get_missing_message(self.param)
+ if msg_extra:
+ if msg:
+ msg += f". {msg_extra}"
+ else:
+ msg = msg_extra
+
+ msg = f" {msg}" if msg else ""
+
+ # Translate param_type for known types.
+ if param_type == "argument":
+ missing = _("Missing argument")
+ elif param_type == "option":
+ missing = _("Missing option")
+ elif param_type == "parameter":
+ missing = _("Missing parameter")
+ else:
+ missing = _("Missing {param_type}").format(param_type=param_type)
+
+ return f"{missing}{param_hint}.{msg}"
+
+ def __str__(self) -> str:
if not self.message:
param_name = self.param.name if self.param else None
- return _('Missing parameter: {param_name}').format(param_name=
- param_name)
+ return _("Missing parameter: {param_name}").format(param_name=param_name)
else:
return self.message
@@ -98,15 +201,32 @@ class NoSuchOption(UsageError):
.. versionadded:: 4.0
"""
- def __init__(self, option_name: str, message: t.Optional[str]=None,
- possibilities: t.Optional[t.Sequence[str]]=None, ctx: t.Optional[
- 'Context']=None) ->None:
+ def __init__(
+ self,
+ option_name: str,
+ message: t.Optional[str] = None,
+ possibilities: t.Optional[t.Sequence[str]] = None,
+ ctx: t.Optional["Context"] = None,
+ ) -> None:
if message is None:
- message = _('No such option: {name}').format(name=option_name)
+ message = _("No such option: {name}").format(name=option_name)
+
super().__init__(message, ctx)
self.option_name = option_name
self.possibilities = possibilities
+ def format_message(self) -> str:
+ if not self.possibilities:
+ return self.message
+
+ possibility_str = ", ".join(sorted(self.possibilities))
+ suggest = ngettext(
+ "Did you mean {possibility}?",
+ "(Possible options: {possibilities})",
+ len(self.possibilities),
+ ).format(possibility=possibility_str, possibilities=possibility_str)
+ return f"{self.message} {suggest}"
+
class BadOptionUsage(UsageError):
"""Raised if an option is generally supplied but the use of the option
@@ -118,8 +238,9 @@ class BadOptionUsage(UsageError):
:param option_name: the name of the option being used incorrectly.
"""
- def __init__(self, option_name: str, message: str, ctx: t.Optional[
- 'Context']=None) ->None:
+ def __init__(
+ self, option_name: str, message: str, ctx: t.Optional["Context"] = None
+ ) -> None:
super().__init__(message, ctx)
self.option_name = option_name
@@ -136,13 +257,19 @@ class BadArgumentUsage(UsageError):
class FileError(ClickException):
"""Raised if a file cannot be opened."""
- def __init__(self, filename: str, hint: t.Optional[str]=None) ->None:
+ def __init__(self, filename: str, hint: t.Optional[str] = None) -> None:
if hint is None:
- hint = _('unknown error')
+ hint = _("unknown error")
+
super().__init__(hint)
self.ui_filename: str = format_filename(filename)
self.filename = filename
+ def format_message(self) -> str:
+ return _("Could not open file {filename!r}: {message}").format(
+ filename=self.ui_filename, message=self.message
+ )
+
class Abort(RuntimeError):
"""An internal signalling exception that signals Click to abort."""
@@ -154,7 +281,8 @@ class Exit(RuntimeError):
:param code: the status code to exit with.
"""
- __slots__ = 'exit_code',
- def __init__(self, code: int=0) ->None:
+ __slots__ = ("exit_code",)
+
+ def __init__(self, code: int = 0) -> None:
self.exit_code: int = code
diff --git a/src/click/formatting.py b/src/click/formatting.py
index 2586652..ddd2a2f 100644
--- a/src/click/formatting.py
+++ b/src/click/formatting.py
@@ -1,13 +1,38 @@
import typing as t
from contextlib import contextmanager
from gettext import gettext as _
+
from ._compat import term_len
from .parser import split_opt
+
+# Can force a width. This is used by the test system
FORCED_WIDTH: t.Optional[int] = None
-def wrap_text(text: str, width: int=78, initial_indent: str='',
- subsequent_indent: str='', preserve_paragraphs: bool=False) ->str:
+def measure_table(rows: t.Iterable[t.Tuple[str, str]]) -> t.Tuple[int, ...]:
+ widths: t.Dict[int, int] = {}
+
+ for row in rows:
+ for idx, col in enumerate(row):
+ widths[idx] = max(widths.get(idx, 0), term_len(col))
+
+ return tuple(y for x, y in sorted(widths.items()))
+
+
+def iter_rows(
+ rows: t.Iterable[t.Tuple[str, str]], col_count: int
+) -> t.Iterator[t.Tuple[str, ...]]:
+ for row in rows:
+ yield row + ("",) * (col_count - len(row))
+
+
+def wrap_text(
+ text: str,
+ width: int = 78,
+ initial_indent: str = "",
+ subsequent_indent: str = "",
+ preserve_paragraphs: bool = False,
+) -> str:
"""A helper function that intelligently wraps text. By default, it
assumes that it operates on a single paragraph of text but if the
`preserve_paragraphs` parameter is provided it will intelligently
@@ -26,7 +51,52 @@ def wrap_text(text: str, width: int=78, initial_indent: str='',
:param preserve_paragraphs: if this flag is set then the wrapping will
intelligently handle paragraphs.
"""
- pass
+ from ._textwrap import TextWrapper
+
+ text = text.expandtabs()
+ wrapper = TextWrapper(
+ width,
+ initial_indent=initial_indent,
+ subsequent_indent=subsequent_indent,
+ replace_whitespace=False,
+ )
+ if not preserve_paragraphs:
+ return wrapper.fill(text)
+
+ p: t.List[t.Tuple[int, bool, str]] = []
+ buf: t.List[str] = []
+ indent = None
+
+ def _flush_par() -> None:
+ if not buf:
+ return
+ if buf[0].strip() == "\b":
+ p.append((indent or 0, True, "\n".join(buf[1:])))
+ else:
+ p.append((indent or 0, False, " ".join(buf)))
+ del buf[:]
+
+ for line in text.splitlines():
+ if not line:
+ _flush_par()
+ indent = None
+ else:
+ if indent is None:
+ orig_len = term_len(line)
+ line = line.lstrip()
+ indent = orig_len - term_len(line)
+ buf.append(line)
+ _flush_par()
+
+ rv = []
+ for indent, raw, text in p:
+ with wrapper.extra_indent(" " * indent):
+ if raw:
+ rv.append(wrapper.indent_only(text))
+ else:
+ rv.append(wrapper.fill(text))
+
+ return "\n\n".join(rv)
class HelpFormatter:
@@ -41,35 +111,40 @@ class HelpFormatter:
width clamped to a maximum of 78.
"""
- def __init__(self, indent_increment: int=2, width: t.Optional[int]=None,
- max_width: t.Optional[int]=None) ->None:
+ def __init__(
+ self,
+ indent_increment: int = 2,
+ width: t.Optional[int] = None,
+ max_width: t.Optional[int] = None,
+ ) -> None:
import shutil
+
self.indent_increment = indent_increment
if max_width is None:
max_width = 80
if width is None:
width = FORCED_WIDTH
if width is None:
- width = max(min(shutil.get_terminal_size().columns,
- max_width) - 2, 50)
+ width = max(min(shutil.get_terminal_size().columns, max_width) - 2, 50)
self.width = width
self.current_indent = 0
self.buffer: t.List[str] = []
- def write(self, string: str) ->None:
+ def write(self, string: str) -> None:
"""Writes a unicode string into the internal buffer."""
- pass
+ self.buffer.append(string)
- def indent(self) ->None:
+ def indent(self) -> None:
"""Increases the indentation."""
- pass
+ self.current_indent += self.indent_increment
- def dedent(self) ->None:
+ def dedent(self) -> None:
"""Decreases the indentation."""
- pass
+ self.current_indent -= self.indent_increment
- def write_usage(self, prog: str, args: str='', prefix: t.Optional[str]=None
- ) ->None:
+ def write_usage(
+ self, prog: str, args: str = "", prefix: t.Optional[str] = None
+ ) -> None:
"""Writes a usage line into the buffer.
:param prog: the program name.
@@ -77,24 +152,67 @@ class HelpFormatter:
:param prefix: The prefix for the first line. Defaults to
``"Usage: "``.
"""
- pass
+ if prefix is None:
+ prefix = f"{_('Usage:')} "
+
+ usage_prefix = f"{prefix:>{self.current_indent}}{prog} "
+ text_width = self.width - self.current_indent
+
+ if text_width >= (term_len(usage_prefix) + 20):
+ # The arguments will fit to the right of the prefix.
+ indent = " " * term_len(usage_prefix)
+ self.write(
+ wrap_text(
+ args,
+ text_width,
+ initial_indent=usage_prefix,
+ subsequent_indent=indent,
+ )
+ )
+ else:
+ # The prefix is too long, put the arguments on the next line.
+ self.write(usage_prefix)
+ self.write("\n")
+ indent = " " * (max(self.current_indent, term_len(prefix)) + 4)
+ self.write(
+ wrap_text(
+ args, text_width, initial_indent=indent, subsequent_indent=indent
+ )
+ )
+
+ self.write("\n")
- def write_heading(self, heading: str) ->None:
+ def write_heading(self, heading: str) -> None:
"""Writes a heading into the buffer."""
- pass
+ self.write(f"{'':>{self.current_indent}}{heading}:\n")
- def write_paragraph(self) ->None:
+ def write_paragraph(self) -> None:
"""Writes a paragraph into the buffer."""
- pass
+ if self.buffer:
+ self.write("\n")
- def write_text(self, text: str) ->None:
+ def write_text(self, text: str) -> None:
"""Writes re-indented text into the buffer. This rewraps and
preserves paragraphs.
"""
- pass
+ indent = " " * self.current_indent
+ self.write(
+ wrap_text(
+ text,
+ self.width,
+ initial_indent=indent,
+ subsequent_indent=indent,
+ preserve_paragraphs=True,
+ )
+ )
+ self.write("\n")
- def write_dl(self, rows: t.Sequence[t.Tuple[str, str]], col_max: int=30,
- col_spacing: int=2) ->None:
+ def write_dl(
+ self,
+ rows: t.Sequence[t.Tuple[str, str]],
+ col_max: int = 30,
+ col_spacing: int = 2,
+ ) -> None:
"""Writes a definition list into the buffer. This is how options
and commands are usually formatted.
@@ -103,31 +221,81 @@ class HelpFormatter:
:param col_spacing: the number of spaces between the first and
second column.
"""
- pass
+ rows = list(rows)
+ widths = measure_table(rows)
+ if len(widths) != 2:
+ raise TypeError("Expected two columns for definition list")
+
+ first_col = min(widths[0], col_max) + col_spacing
+
+ for first, second in iter_rows(rows, len(widths)):
+ self.write(f"{'':>{self.current_indent}}{first}")
+ if not second:
+ self.write("\n")
+ continue
+ if term_len(first) <= first_col - col_spacing:
+ self.write(" " * (first_col - term_len(first)))
+ else:
+ self.write("\n")
+ self.write(" " * (first_col + self.current_indent))
+
+ text_width = max(self.width - first_col - 2, 10)
+ wrapped_text = wrap_text(second, text_width, preserve_paragraphs=True)
+ lines = wrapped_text.splitlines()
+
+ if lines:
+ self.write(f"{lines[0]}\n")
+
+ for line in lines[1:]:
+ self.write(f"{'':>{first_col + self.current_indent}}{line}\n")
+ else:
+ self.write("\n")
@contextmanager
- def section(self, name: str) ->t.Iterator[None]:
+ def section(self, name: str) -> t.Iterator[None]:
"""Helpful context manager that writes a paragraph, a heading,
and the indents.
:param name: the section name that is written as heading.
"""
- pass
+ self.write_paragraph()
+ self.write_heading(name)
+ self.indent()
+ try:
+ yield
+ finally:
+ self.dedent()
@contextmanager
- def indentation(self) ->t.Iterator[None]:
+ def indentation(self) -> t.Iterator[None]:
"""A context manager that increases the indentation."""
- pass
+ self.indent()
+ try:
+ yield
+ finally:
+ self.dedent()
- def getvalue(self) ->str:
+ def getvalue(self) -> str:
"""Returns the buffer contents."""
- pass
+ return "".join(self.buffer)
-def join_options(options: t.Sequence[str]) ->t.Tuple[str, bool]:
+def join_options(options: t.Sequence[str]) -> t.Tuple[str, bool]:
"""Given a list of option strings this joins them in the most appropriate
way and returns them in the form ``(formatted_string,
any_prefix_is_slash)`` where the second item in the tuple is a flag that
indicates if any of the option prefixes was a slash.
"""
- pass
+ rv = []
+ any_prefix_is_slash = False
+
+ for opt in options:
+ prefix = split_opt(opt)[0]
+
+ if prefix == "/":
+ any_prefix_is_slash = True
+
+ rv.append((len(prefix), opt))
+
+ rv.sort(key=lambda x: x[0])
+ return ", ".join(x[1] for x in rv), any_prefix_is_slash
diff --git a/src/click/globals.py b/src/click/globals.py
index ca86d44..480058f 100644
--- a/src/click/globals.py
+++ b/src/click/globals.py
@@ -1,12 +1,24 @@
import typing as t
from threading import local
+
if t.TYPE_CHECKING:
import typing_extensions as te
from .core import Context
+
_local = local()
-def get_current_context(silent: bool=False) ->t.Optional['Context']:
+@t.overload
+def get_current_context(silent: "te.Literal[False]" = False) -> "Context":
+ ...
+
+
+@t.overload
+def get_current_context(silent: bool = ...) -> t.Optional["Context"]:
+ ...
+
+
+def get_current_context(silent: bool = False) -> t.Optional["Context"]:
"""Returns the current click context. This can be used as a way to
access the current context object from anywhere. This is a more implicit
alternative to the :func:`pass_context` decorator. This function is
@@ -21,22 +33,36 @@ def get_current_context(silent: bool=False) ->t.Optional['Context']:
is available. The default behavior is to raise a
:exc:`RuntimeError`.
"""
- pass
+ try:
+ return t.cast("Context", _local.stack[-1])
+ except (AttributeError, IndexError) as e:
+ if not silent:
+ raise RuntimeError("There is no active click context.") from e
+
+ return None
-def push_context(ctx: 'Context') ->None:
+def push_context(ctx: "Context") -> None:
"""Pushes a new context to the current stack."""
- pass
+ _local.__dict__.setdefault("stack", []).append(ctx)
-def pop_context() ->None:
+def pop_context() -> None:
"""Removes the top level from the stack."""
- pass
+ _local.stack.pop()
-def resolve_color_default(color: t.Optional[bool]=None) ->t.Optional[bool]:
+def resolve_color_default(color: t.Optional[bool] = None) -> t.Optional[bool]:
"""Internal helper to get the default value of the color flag. If a
value is passed it's returned unchanged, otherwise it's looked up from
the current context.
"""
- pass
+ if color is not None:
+ return color
+
+ ctx = get_current_context(silent=True)
+
+ if ctx is not None:
+ return ctx.color
+
+ return None
diff --git a/src/click/parser.py b/src/click/parser.py
index 5baffed..5fa7adf 100644
--- a/src/click/parser.py
+++ b/src/click/parser.py
@@ -17,26 +17,38 @@ by the Python Software Foundation. This is limited to code in parser.py.
Copyright 2001-2006 Gregory P. Ward. All rights reserved.
Copyright 2002-2006 Python Software Foundation. All rights reserved.
"""
+# This code uses parts of optparse written by Gregory P. Ward and
+# maintained by the Python Software Foundation.
+# Copyright 2001-2006 Gregory P. Ward
+# Copyright 2002-2006 Python Software Foundation
import typing as t
from collections import deque
from gettext import gettext as _
from gettext import ngettext
+
from .exceptions import BadArgumentUsage
from .exceptions import BadOptionUsage
from .exceptions import NoSuchOption
from .exceptions import UsageError
+
if t.TYPE_CHECKING:
import typing_extensions as te
from .core import Argument as CoreArgument
from .core import Context
from .core import Option as CoreOption
from .core import Parameter as CoreParameter
-V = t.TypeVar('V')
+
+V = t.TypeVar("V")
+
+# Sentinel value that indicates 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.
_flag_needs_value = object()
-def _unpack_args(args: t.Sequence[str], nargs_spec: t.Sequence[int]) ->t.Tuple[
- t.Sequence[t.Union[str, t.Sequence[t.Optional[str]], None]], t.List[str]]:
+def _unpack_args(
+ args: t.Sequence[str], nargs_spec: t.Sequence[int]
+) -> t.Tuple[t.Sequence[t.Union[str, t.Sequence[t.Optional[str]], None]], t.List[str]]:
"""Given an iterable of arguments and an iterable of nargs specifications,
it returns a tuple with all the unpacked arguments at the first index
and all remaining arguments as the second.
@@ -46,10 +58,71 @@ def _unpack_args(args: t.Sequence[str], nargs_spec: t.Sequence[int]) ->t.Tuple[
Missing items are filled with `None`.
"""
- pass
+ args = deque(args)
+ nargs_spec = deque(nargs_spec)
+ rv: t.List[t.Union[str, t.Tuple[t.Optional[str], ...], None]] = []
+ spos: t.Optional[int] = None
+
+ def _fetch(c: "te.Deque[V]") -> t.Optional[V]:
+ try:
+ if spos is None:
+ return c.popleft()
+ else:
+ return c.pop()
+ except IndexError:
+ return None
+
+ while nargs_spec:
+ nargs = _fetch(nargs_spec)
+
+ if nargs is None:
+ continue
+
+ if nargs == 1:
+ rv.append(_fetch(args))
+ elif nargs > 1:
+ x = [_fetch(args) for _ in range(nargs)]
+
+ # If we're reversed, we're pulling in the arguments in reverse,
+ # so we need to turn them around.
+ if spos is not None:
+ x.reverse()
+
+ rv.append(tuple(x))
+ elif nargs < 0:
+ if spos is not None:
+ raise TypeError("Cannot have two nargs < 0")
+
+ spos = len(rv)
+ rv.append(None)
+ # spos is the position of the wildcard (star). If it's not `None`,
+ # we fill it with the remainder.
+ if spos is not None:
+ rv[spos] = tuple(args)
+ args = []
+ rv[spos + 1 :] = reversed(rv[spos + 1 :])
-def split_arg_string(string: str) ->t.List[str]:
+ return tuple(rv), list(args)
+
+
+def split_opt(opt: str) -> t.Tuple[str, str]:
+ first = opt[:1]
+ if first.isalnum():
+ return "", opt
+ if opt[1:2] == first:
+ return opt[:2], opt[2:]
+ return first, opt[1:]
+
+
+def normalize_opt(opt: str, ctx: t.Optional["Context"]) -> str:
+ if ctx is None or ctx.token_normalize_func is None:
+ return opt
+ prefix, opt = split_opt(opt)
+ return f"{prefix}{ctx.token_normalize_func(opt)}"
+
+
+def split_arg_string(string: str) -> t.List[str]:
"""Split an argument string as with :func:`shlex.split`, but don't
fail if the string is incomplete. Ignores a missing closing quote or
incomplete escape sequence and uses the partial token as-is.
@@ -64,52 +137,117 @@ def split_arg_string(string: str) ->t.List[str]:
:param string: String to split.
"""
- pass
+ import shlex
+ lex = shlex.shlex(string, posix=True)
+ lex.whitespace_split = True
+ lex.commenters = ""
+ out = []
-class Option:
+ try:
+ for token in lex:
+ out.append(token)
+ except ValueError:
+ # Raised when end-of-string is reached in an invalid state. Use
+ # the partial token as-is. The quote or escape character is in
+ # lex.state, not lex.token.
+ out.append(lex.token)
+
+ return out
- def __init__(self, obj: 'CoreOption', opts: t.Sequence[str], dest: t.
- Optional[str], action: t.Optional[str]=None, nargs: int=1, const: t
- .Optional[t.Any]=None):
+
+class Option:
+ def __init__(
+ self,
+ obj: "CoreOption",
+ opts: t.Sequence[str],
+ dest: t.Optional[str],
+ action: t.Optional[str] = None,
+ nargs: int = 1,
+ const: t.Optional[t.Any] = None,
+ ):
self._short_opts = []
self._long_opts = []
self.prefixes: t.Set[str] = set()
+
for opt in opts:
prefix, value = split_opt(opt)
if not prefix:
- raise ValueError(f'Invalid start character for option ({opt})')
+ raise ValueError(f"Invalid start character for option ({opt})")
self.prefixes.add(prefix[0])
if len(prefix) == 1 and len(value) == 1:
self._short_opts.append(opt)
else:
self._long_opts.append(opt)
self.prefixes.add(prefix)
+
if action is None:
- action = 'store'
+ action = "store"
+
self.dest = dest
self.action = action
self.nargs = nargs
self.const = const
self.obj = obj
+ @property
+ def takes_value(self) -> bool:
+ return self.action in ("store", "append")
+
+ def process(self, value: t.Any, state: "ParsingState") -> None:
+ if self.action == "store":
+ state.opts[self.dest] = value # type: ignore
+ elif self.action == "store_const":
+ state.opts[self.dest] = self.const # type: ignore
+ elif self.action == "append":
+ state.opts.setdefault(self.dest, []).append(value) # type: ignore
+ elif self.action == "append_const":
+ state.opts.setdefault(self.dest, []).append(self.const) # type: ignore
+ elif self.action == "count":
+ state.opts[self.dest] = state.opts.get(self.dest, 0) + 1 # type: ignore
+ else:
+ raise ValueError(f"unknown action '{self.action}'")
+ state.order.append(self.obj)
-class Argument:
- def __init__(self, obj: 'CoreArgument', dest: t.Optional[str], nargs: int=1
- ):
+class Argument:
+ def __init__(self, obj: "CoreArgument", dest: t.Optional[str], nargs: int = 1):
self.dest = dest
self.nargs = nargs
self.obj = obj
+ def process(
+ self,
+ value: t.Union[t.Optional[str], t.Sequence[t.Optional[str]]],
+ state: "ParsingState",
+ ) -> None:
+ if self.nargs > 1:
+ assert value is not None
+ holes = sum(1 for x in value if x is None)
+ if holes == len(value):
+ value = None
+ elif holes != 0:
+ raise BadArgumentUsage(
+ _("Argument {name!r} takes {nargs} values.").format(
+ name=self.dest, nargs=self.nargs
+ )
+ )
+
+ if self.nargs == -1 and self.obj.envvar is not None and value == ():
+ # Replace empty tuple with None so that a value from the
+ # environment may be tried.
+ value = None
+
+ state.opts[self.dest] = value # type: ignore
+ state.order.append(self.obj)
-class ParsingState:
- def __init__(self, rargs: t.List[str]) ->None:
+class ParsingState:
+ def __init__(self, rargs: t.List[str]) -> None:
self.opts: t.Dict[str, t.Any] = {}
self.largs: t.List[str] = []
self.rargs = rargs
- self.order: t.List['CoreParameter'] = []
+ self.order: t.List["CoreParameter"] = []
class OptionParser:
@@ -126,21 +264,39 @@ class OptionParser:
should go with.
"""
- def __init__(self, ctx: t.Optional['Context']=None) ->None:
+ def __init__(self, ctx: t.Optional["Context"] = None) -> None:
+ #: The :class:`~click.Context` for this parser. This might be
+ #: `None` for some advanced use cases.
self.ctx = ctx
+ #: This controls how the parser deals with interspersed arguments.
+ #: If this is set to `False`, the parser will stop on the first
+ #: non-option. Click uses this to implement nested subcommands
+ #: safely.
self.allow_interspersed_args: bool = True
+ #: This tells the parser how to deal with unknown options. By
+ #: default it will error out (which is sensible), but there is a
+ #: second mode where it will ignore it and continue processing
+ #: after shifting all the unknown options into the resulting args.
self.ignore_unknown_options: bool = False
+
if ctx is not None:
self.allow_interspersed_args = ctx.allow_interspersed_args
self.ignore_unknown_options = ctx.ignore_unknown_options
+
self._short_opt: t.Dict[str, Option] = {}
self._long_opt: t.Dict[str, Option] = {}
- self._opt_prefixes = {'-', '--'}
+ self._opt_prefixes = {"-", "--"}
self._args: t.List[Argument] = []
- def add_option(self, obj: 'CoreOption', opts: t.Sequence[str], dest: t.
- Optional[str], action: t.Optional[str]=None, nargs: int=1, const: t
- .Optional[t.Any]=None) ->None:
+ def add_option(
+ self,
+ obj: "CoreOption",
+ opts: t.Sequence[str],
+ dest: t.Optional[str],
+ action: t.Optional[str] = None,
+ nargs: int = 1,
+ const: t.Optional[t.Any] = None,
+ ) -> None:
"""Adds a new option named `dest` to the parser. The destination
is not inferred (unlike with optparse) and needs to be explicitly
provided. Action can be any of ``store``, ``store_const``,
@@ -149,23 +305,225 @@ class OptionParser:
The `obj` can be used to identify the option in the order list
that is returned from the parser.
"""
- pass
-
- def add_argument(self, obj: 'CoreArgument', dest: t.Optional[str],
- nargs: int=1) ->None:
+ opts = [normalize_opt(opt, self.ctx) for opt in opts]
+ option = Option(obj, opts, dest, action=action, nargs=nargs, const=const)
+ self._opt_prefixes.update(option.prefixes)
+ for opt in option._short_opts:
+ self._short_opt[opt] = option
+ for opt in option._long_opts:
+ self._long_opt[opt] = option
+
+ def add_argument(
+ self, obj: "CoreArgument", dest: t.Optional[str], nargs: int = 1
+ ) -> None:
"""Adds a positional argument named `dest` to the parser.
The `obj` can be used to identify the option in the order list
that is returned from the parser.
"""
- pass
+ self._args.append(Argument(obj, dest=dest, nargs=nargs))
- def parse_args(self, args: t.List[str]) ->t.Tuple[t.Dict[str, t.Any], t
- .List[str], t.List['CoreParameter']]:
+ def parse_args(
+ self, args: t.List[str]
+ ) -> t.Tuple[t.Dict[str, t.Any], t.List[str], t.List["CoreParameter"]]:
"""Parses positional arguments and returns ``(values, args, order)``
for the parsed options and arguments as well as the leftover
arguments if there are any. The order is a list of objects as they
appear on the command line. If arguments appear multiple times they
will be memorized multiple times as well.
"""
- pass
+ state = ParsingState(args)
+ try:
+ self._process_args_for_options(state)
+ self._process_args_for_args(state)
+ except UsageError:
+ if self.ctx is None or not self.ctx.resilient_parsing:
+ raise
+ return state.opts, state.largs, state.order
+
+ def _process_args_for_args(self, state: ParsingState) -> None:
+ pargs, args = _unpack_args(
+ state.largs + state.rargs, [x.nargs for x in self._args]
+ )
+
+ for idx, arg in enumerate(self._args):
+ arg.process(pargs[idx], state)
+
+ state.largs = args
+ state.rargs = []
+
+ def _process_args_for_options(self, state: ParsingState) -> None:
+ while state.rargs:
+ arg = state.rargs.pop(0)
+ arglen = len(arg)
+ # Double dashes always handled explicitly regardless of what
+ # prefixes are valid.
+ if arg == "--":
+ return
+ elif arg[:1] in self._opt_prefixes and arglen > 1:
+ self._process_opts(arg, state)
+ elif self.allow_interspersed_args:
+ state.largs.append(arg)
+ else:
+ state.rargs.insert(0, arg)
+ return
+
+ # Say this is the original argument list:
+ # [arg0, arg1, ..., arg(i-1), arg(i), arg(i+1), ..., arg(N-1)]
+ # ^
+ # (we are about to process arg(i)).
+ #
+ # Then rargs is [arg(i), ..., arg(N-1)] and largs is a *subset* of
+ # [arg0, ..., arg(i-1)] (any options and their arguments will have
+ # been removed from largs).
+ #
+ # The while loop will usually consume 1 or more arguments per pass.
+ # If it consumes 1 (eg. arg is an option that takes no arguments),
+ # then after _process_arg() is done the situation is:
+ #
+ # largs = subset of [arg0, ..., arg(i)]
+ # rargs = [arg(i+1), ..., arg(N-1)]
+ #
+ # If allow_interspersed_args is false, largs will always be
+ # *empty* -- still a subset of [arg0, ..., arg(i-1)], but
+ # not a very interesting subset!
+
+ def _match_long_opt(
+ self, opt: str, explicit_value: t.Optional[str], state: ParsingState
+ ) -> None:
+ if opt not in self._long_opt:
+ from difflib import get_close_matches
+
+ possibilities = get_close_matches(opt, self._long_opt)
+ raise NoSuchOption(opt, possibilities=possibilities, ctx=self.ctx)
+
+ option = self._long_opt[opt]
+ if option.takes_value:
+ # At this point it's safe to modify rargs by injecting the
+ # explicit value, because no exception is raised in this
+ # branch. This means that the inserted value will be fully
+ # consumed.
+ if explicit_value is not None:
+ state.rargs.insert(0, explicit_value)
+
+ value = self._get_value_from_state(opt, option, state)
+
+ elif explicit_value is not None:
+ raise BadOptionUsage(
+ opt, _("Option {name!r} does not take a value.").format(name=opt)
+ )
+
+ else:
+ value = None
+
+ option.process(value, state)
+
+ def _match_short_opt(self, arg: str, state: ParsingState) -> None:
+ stop = False
+ i = 1
+ prefix = arg[0]
+ unknown_options = []
+
+ for ch in arg[1:]:
+ opt = normalize_opt(f"{prefix}{ch}", self.ctx)
+ option = self._short_opt.get(opt)
+ i += 1
+
+ if not option:
+ if self.ignore_unknown_options:
+ unknown_options.append(ch)
+ continue
+ raise NoSuchOption(opt, ctx=self.ctx)
+ if option.takes_value:
+ # Any characters left in arg? Pretend they're the
+ # next arg, and stop consuming characters of arg.
+ if i < len(arg):
+ state.rargs.insert(0, arg[i:])
+ stop = True
+
+ value = self._get_value_from_state(opt, option, state)
+
+ else:
+ value = None
+
+ option.process(value, state)
+
+ if stop:
+ break
+
+ # If we got any unknown options we recombine the string of the
+ # remaining options and re-attach the prefix, then report that
+ # to the state as new larg. This way there is basic combinatorics
+ # that can be achieved while still ignoring unknown arguments.
+ if self.ignore_unknown_options and unknown_options:
+ state.largs.append(f"{prefix}{''.join(unknown_options)}")
+
+ def _get_value_from_state(
+ self, option_name: str, option: Option, state: ParsingState
+ ) -> t.Any:
+ nargs = option.nargs
+
+ if len(state.rargs) < nargs:
+ if option.obj._flag_needs_value:
+ # Option allows omitting the value.
+ value = _flag_needs_value
+ else:
+ raise BadOptionUsage(
+ option_name,
+ ngettext(
+ "Option {name!r} requires an argument.",
+ "Option {name!r} requires {nargs} arguments.",
+ nargs,
+ ).format(name=option_name, nargs=nargs),
+ )
+ elif nargs == 1:
+ next_rarg = state.rargs[0]
+
+ if (
+ option.obj._flag_needs_value
+ and isinstance(next_rarg, str)
+ and next_rarg[:1] in self._opt_prefixes
+ and len(next_rarg) > 1
+ ):
+ # The next arg looks like the start of an option, don't
+ # use it as the value if omitting the value is allowed.
+ value = _flag_needs_value
+ else:
+ value = state.rargs.pop(0)
+ else:
+ value = tuple(state.rargs[:nargs])
+ del state.rargs[:nargs]
+
+ return value
+
+ def _process_opts(self, arg: str, state: ParsingState) -> None:
+ explicit_value = None
+ # Long option handling happens in two parts. The first part is
+ # supporting explicitly attached values. In any case, we will try
+ # to long match the option first.
+ if "=" in arg:
+ long_opt, explicit_value = arg.split("=", 1)
+ else:
+ long_opt = arg
+ norm_long_opt = normalize_opt(long_opt, self.ctx)
+
+ # At this point we will match the (assumed) long option through
+ # the long option matching code. Note that this allows options
+ # like "-foo" to be matched as long options.
+ try:
+ self._match_long_opt(norm_long_opt, explicit_value, state)
+ except NoSuchOption:
+ # At this point the long option matching failed, and we need
+ # to try with short options. However there is a special rule
+ # which says, that if we have a two character options prefix
+ # (applies to "--foo" for instance), we do not dispatch to the
+ # short option code and will instead raise the no option
+ # error.
+ if arg[:2] not in self._opt_prefixes:
+ self._match_short_opt(arg, state)
+ return
+
+ if not self.ignore_unknown_options:
+ raise
+
+ state.largs.append(arg)
diff --git a/src/click/shell_completion.py b/src/click/shell_completion.py
index 6956829..dc9e00b 100644
--- a/src/click/shell_completion.py
+++ b/src/click/shell_completion.py
@@ -2,6 +2,7 @@ import os
import re
import typing as t
from gettext import gettext as _
+
from .core import Argument
from .core import BaseCommand
from .core import Context
@@ -13,8 +14,13 @@ from .parser import split_arg_string
from .utils import echo
-def shell_complete(cli: BaseCommand, ctx_args: t.MutableMapping[str, t.Any],
- prog_name: str, complete_var: str, instruction: str) ->int:
+def shell_complete(
+ cli: BaseCommand,
+ ctx_args: t.MutableMapping[str, t.Any],
+ prog_name: str,
+ complete_var: str,
+ instruction: str,
+) -> int:
"""Perform shell completion for the given CLI program.
:param cli: Command being called.
@@ -27,7 +33,23 @@ def shell_complete(cli: BaseCommand, ctx_args: t.MutableMapping[str, t.Any],
instruction and shell, in the form ``instruction_shell``.
:return: Status code to exit with.
"""
- pass
+ shell, _, instruction = instruction.partition("_")
+ comp_cls = get_completion_class(shell)
+
+ if comp_cls is None:
+ return 1
+
+ comp = comp_cls(cli, ctx_args, prog_name, complete_var)
+
+ if instruction == "source":
+ echo(comp.source())
+ return 0
+
+ if instruction == "complete":
+ echo(comp.complete())
+ return 0
+
+ return 1
class CompletionItem:
@@ -48,24 +70,33 @@ class CompletionItem:
don't use this, but custom type completions paired with custom
shell support could use it.
"""
- __slots__ = 'value', 'type', 'help', '_info'
- def __init__(self, value: t.Any, type: str='plain', help: t.Optional[
- str]=None, **kwargs: t.Any) ->None:
+ __slots__ = ("value", "type", "help", "_info")
+
+ def __init__(
+ self,
+ value: t.Any,
+ type: str = "plain",
+ help: t.Optional[str] = None,
+ **kwargs: t.Any,
+ ) -> None:
self.value: t.Any = value
self.type: str = type
self.help: t.Optional[str] = help
self._info = kwargs
- def __getattr__(self, name: str) ->t.Any:
+ def __getattr__(self, name: str) -> t.Any:
return self._info.get(name)
-_SOURCE_BASH = """%(complete_func)s() {
+# Only Bash >= 4.4 has the nosort option.
+_SOURCE_BASH = """\
+%(complete_func)s() {
local IFS=$'\\n'
local response
- response=$(env COMP_WORDS="${COMP_WORDS[*]}" COMP_CWORD=$COMP_CWORD %(complete_var)s=bash_complete $1)
+ response=$(env COMP_WORDS="${COMP_WORDS[*]}" COMP_CWORD=$COMP_CWORD \
+%(complete_var)s=bash_complete $1)
for completion in $response; do
IFS=',' read type value <<< "$completion"
@@ -90,7 +121,9 @@ _SOURCE_BASH = """%(complete_func)s() {
%(complete_func)s_setup;
"""
-_SOURCE_ZSH = """#compdef %(prog_name)s
+
+_SOURCE_ZSH = """\
+#compdef %(prog_name)s
%(complete_func)s() {
local -a completions
@@ -98,7 +131,8 @@ _SOURCE_ZSH = """#compdef %(prog_name)s
local -a response
(( ! $+commands[%(prog_name)s] )) && return 1
- response=("${(@f)$(env COMP_WORDS="${words[*]}" COMP_CWORD=$((CURRENT-1)) %(complete_var)s=zsh_complete %(prog_name)s)}")
+ response=("${(@f)$(env COMP_WORDS="${words[*]}" COMP_CWORD=$((CURRENT-1)) \
+%(complete_var)s=zsh_complete %(prog_name)s)}")
for type key descr in ${response}; do
if [[ "$type" == "plain" ]]; then
@@ -131,8 +165,11 @@ else
compdef %(complete_func)s %(prog_name)s
fi
"""
-_SOURCE_FISH = """function %(complete_func)s;
- set -l response (env %(complete_var)s=fish_complete COMP_WORDS=(commandline -cp) COMP_CWORD=(commandline -t) %(prog_name)s);
+
+_SOURCE_FISH = """\
+function %(complete_func)s;
+ set -l response (env %(complete_var)s=fish_complete COMP_WORDS=(commandline -cp) \
+COMP_CWORD=(commandline -t) %(prog_name)s);
for completion in $response;
set -l metadata (string split "," $completion);
@@ -147,7 +184,8 @@ _SOURCE_FISH = """function %(complete_func)s;
end;
end;
-complete --no-files --command %(prog_name)s --arguments "(%(complete_func)s)";
+complete --no-files --command %(prog_name)s --arguments \
+"(%(complete_func)s)";
"""
@@ -163,55 +201,68 @@ class ShellComplete:
.. versionadded:: 8.0
"""
+
name: t.ClassVar[str]
"""Name to register the shell as with :func:`add_completion_class`.
This is used in completion instructions (``{name}_source`` and
``{name}_complete``).
"""
+
source_template: t.ClassVar[str]
"""Completion script template formatted by :meth:`source`. This must
be provided by subclasses.
"""
- def __init__(self, cli: BaseCommand, ctx_args: t.MutableMapping[str, t.
- Any], prog_name: str, complete_var: str) ->None:
+ def __init__(
+ self,
+ cli: BaseCommand,
+ ctx_args: t.MutableMapping[str, t.Any],
+ prog_name: str,
+ complete_var: str,
+ ) -> None:
self.cli = cli
self.ctx_args = ctx_args
self.prog_name = prog_name
self.complete_var = complete_var
@property
- def func_name(self) ->str:
+ def func_name(self) -> str:
"""The name of the shell function defined by the completion
script.
"""
- pass
+ safe_name = re.sub(r"\W*", "", self.prog_name.replace("-", "_"), flags=re.ASCII)
+ return f"_{safe_name}_completion"
- def source_vars(self) ->t.Dict[str, t.Any]:
+ def source_vars(self) -> t.Dict[str, t.Any]:
"""Vars for formatting :attr:`source_template`.
By default this provides ``complete_func``, ``complete_var``,
and ``prog_name``.
"""
- pass
+ return {
+ "complete_func": self.func_name,
+ "complete_var": self.complete_var,
+ "prog_name": self.prog_name,
+ }
- def source(self) ->str:
+ def source(self) -> str:
"""Produce the shell script that defines the completion
function. By default this ``%``-style formats
:attr:`source_template` with the dict returned by
:meth:`source_vars`.
"""
- pass
+ return self.source_template % self.source_vars()
- def get_completion_args(self) ->t.Tuple[t.List[str], str]:
+ def get_completion_args(self) -> t.Tuple[t.List[str], str]:
"""Use the env vars defined by the shell script to return a
tuple of ``args, incomplete``. This must be implemented by
subclasses.
"""
- pass
+ raise NotImplementedError
- def get_completions(self, args: t.List[str], incomplete: str) ->t.List[
- CompletionItem]:
+ def get_completions(
+ self, args: t.List[str], incomplete: str
+ ) -> t.List[CompletionItem]:
"""Determine the context and last complete command or parameter
from the complete args. Call that object's ``shell_complete``
method to get the completions for the incomplete value.
@@ -219,51 +270,143 @@ class ShellComplete:
:param args: List of complete args before the incomplete value.
:param incomplete: Value being completed. May be empty.
"""
- pass
+ ctx = _resolve_context(self.cli, self.ctx_args, self.prog_name, args)
+ obj, incomplete = _resolve_incomplete(ctx, args, incomplete)
+ return obj.shell_complete(ctx, incomplete)
- def format_completion(self, item: CompletionItem) ->str:
+ def format_completion(self, item: CompletionItem) -> str:
"""Format a completion item into the form recognized by the
shell script. This must be implemented by subclasses.
:param item: Completion item to format.
"""
- pass
+ raise NotImplementedError
- def complete(self) ->str:
+ def complete(self) -> str:
"""Produce the completion data to send back to the shell.
By default this calls :meth:`get_completion_args`, gets the
completions, then calls :meth:`format_completion` for each
completion.
"""
- pass
+ args, incomplete = self.get_completion_args()
+ completions = self.get_completions(args, incomplete)
+ out = [self.format_completion(item) for item in completions]
+ return "\n".join(out)
class BashComplete(ShellComplete):
"""Shell completion for Bash."""
- name = 'bash'
+
+ name = "bash"
source_template = _SOURCE_BASH
+ @staticmethod
+ def _check_version() -> None:
+ import subprocess
+
+ output = subprocess.run(
+ ["bash", "-c", 'echo "${BASH_VERSION}"'], stdout=subprocess.PIPE
+ )
+ match = re.search(r"^(\d+)\.(\d+)\.\d+", output.stdout.decode())
+
+ if match is not None:
+ major, minor = match.groups()
+
+ if major < "4" or major == "4" and minor < "4":
+ echo(
+ _(
+ "Shell completion is not supported for Bash"
+ " versions older than 4.4."
+ ),
+ err=True,
+ )
+ else:
+ echo(
+ _("Couldn't detect Bash version, shell completion is not supported."),
+ err=True,
+ )
+
+ def source(self) -> str:
+ self._check_version()
+ return super().source()
+
+ def get_completion_args(self) -> t.Tuple[t.List[str], str]:
+ cwords = split_arg_string(os.environ["COMP_WORDS"])
+ cword = int(os.environ["COMP_CWORD"])
+ args = cwords[1:cword]
+
+ try:
+ incomplete = cwords[cword]
+ except IndexError:
+ incomplete = ""
+
+ return args, incomplete
+
+ def format_completion(self, item: CompletionItem) -> str:
+ return f"{item.type},{item.value}"
+
class ZshComplete(ShellComplete):
"""Shell completion for Zsh."""
- name = 'zsh'
+
+ name = "zsh"
source_template = _SOURCE_ZSH
+ def get_completion_args(self) -> t.Tuple[t.List[str], str]:
+ cwords = split_arg_string(os.environ["COMP_WORDS"])
+ cword = int(os.environ["COMP_CWORD"])
+ args = cwords[1:cword]
+
+ try:
+ incomplete = cwords[cword]
+ except IndexError:
+ incomplete = ""
+
+ return args, incomplete
+
+ def format_completion(self, item: CompletionItem) -> str:
+ return f"{item.type}\n{item.value}\n{item.help if item.help else '_'}"
+
class FishComplete(ShellComplete):
"""Shell completion for Fish."""
- name = 'fish'
+
+ name = "fish"
source_template = _SOURCE_FISH
+ def get_completion_args(self) -> t.Tuple[t.List[str], str]:
+ cwords = split_arg_string(os.environ["COMP_WORDS"])
+ incomplete = os.environ["COMP_CWORD"]
+ args = cwords[1:]
-ShellCompleteType = t.TypeVar('ShellCompleteType', bound=t.Type[ShellComplete])
-_available_shells: t.Dict[str, t.Type[ShellComplete]] = {'bash':
- BashComplete, 'fish': FishComplete, 'zsh': ZshComplete}
+ # Fish stores the partial word in both COMP_WORDS and
+ # COMP_CWORD, remove it from complete args.
+ if incomplete and args and args[-1] == incomplete:
+ args.pop()
+ return args, incomplete
-def add_completion_class(cls: ShellCompleteType, name: t.Optional[str]=None
- ) ->ShellCompleteType:
+ def format_completion(self, item: CompletionItem) -> str:
+ if item.help:
+ return f"{item.type},{item.value}\t{item.help}"
+
+ return f"{item.type},{item.value}"
+
+
+ShellCompleteType = t.TypeVar("ShellCompleteType", bound=t.Type[ShellComplete])
+
+
+_available_shells: t.Dict[str, t.Type[ShellComplete]] = {
+ "bash": BashComplete,
+ "fish": FishComplete,
+ "zsh": ZshComplete,
+}
+
+
+def add_completion_class(
+ cls: ShellCompleteType, name: t.Optional[str] = None
+) -> ShellCompleteType:
"""Register a :class:`ShellComplete` subclass under the given name.
The name will be provided by the completion instruction environment
variable during completion.
@@ -273,20 +416,25 @@ def add_completion_class(cls: ShellCompleteType, name: t.Optional[str]=None
:param name: Name to register the class under. Defaults to the
class's ``name`` attribute.
"""
- pass
+ if name is None:
+ name = cls.name
+
+ _available_shells[name] = cls
+ return cls
-def get_completion_class(shell: str) ->t.Optional[t.Type[ShellComplete]]:
+
+def get_completion_class(shell: str) -> t.Optional[t.Type[ShellComplete]]:
"""Look up a registered :class:`ShellComplete` subclass by the name
provided by the completion instruction environment variable. If the
name isn't registered, returns ``None``.
:param shell: Name the class is registered under.
"""
- pass
+ return _available_shells.get(shell)
-def _is_incomplete_argument(ctx: Context, param: Parameter) ->bool:
+def _is_incomplete_argument(ctx: Context, param: Parameter) -> bool:
"""Determine if the given parameter is an argument that can still
accept values.
@@ -294,26 +442,62 @@ def _is_incomplete_argument(ctx: Context, param: Parameter) ->bool:
parsed complete args.
:param param: Argument object being checked.
"""
- pass
-
-
-def _start_of_option(ctx: Context, value: str) ->bool:
+ if not isinstance(param, Argument):
+ return False
+
+ assert param.name is not None
+ # Will be None if expose_value is False.
+ value = ctx.params.get(param.name)
+ return (
+ param.nargs == -1
+ or ctx.get_parameter_source(param.name) is not ParameterSource.COMMANDLINE
+ or (
+ param.nargs > 1
+ and isinstance(value, (tuple, list))
+ and len(value) < param.nargs
+ )
+ )
+
+
+def _start_of_option(ctx: Context, value: str) -> bool:
"""Check if the value looks like the start of an option."""
- pass
+ if not value:
+ return False
+
+ c = value[0]
+ return c in ctx._opt_prefixes
-def _is_incomplete_option(ctx: Context, args: t.List[str], param: Parameter
- ) ->bool:
+def _is_incomplete_option(ctx: Context, args: t.List[str], param: Parameter) -> bool:
"""Determine if the given parameter is an option that needs a value.
:param args: List of complete args before the incomplete value.
:param param: Option object being checked.
"""
- pass
+ if not isinstance(param, Option):
+ return False
+
+ if param.is_flag or param.count:
+ return False
+ last_option = None
-def _resolve_context(cli: BaseCommand, ctx_args: t.MutableMapping[str, t.
- Any], prog_name: str, args: t.List[str]) ->Context:
+ for index, arg in enumerate(reversed(args)):
+ if index + 1 > param.nargs:
+ break
+
+ if _start_of_option(ctx, arg):
+ last_option = arg
+
+ return last_option is not None and last_option in param.opts
+
+
+def _resolve_context(
+ cli: BaseCommand,
+ ctx_args: t.MutableMapping[str, t.Any],
+ prog_name: str,
+ args: t.List[str],
+) -> Context:
"""Produce the context hierarchy starting with the command and
traversing the complete arguments. This only follows the commands,
it doesn't trigger input prompts or callbacks.
@@ -322,11 +506,52 @@ def _resolve_context(cli: BaseCommand, ctx_args: t.MutableMapping[str, t.
:param prog_name: Name of the executable in the shell.
:param args: List of complete args before the incomplete value.
"""
- pass
+ ctx_args["resilient_parsing"] = True
+ ctx = cli.make_context(prog_name, args.copy(), **ctx_args)
+ args = ctx.protected_args + ctx.args
+
+ while args:
+ command = ctx.command
+
+ if isinstance(command, MultiCommand):
+ if not command.chain:
+ name, cmd, args = command.resolve_command(ctx, args)
+
+ if cmd is None:
+ return ctx
+
+ ctx = cmd.make_context(name, args, parent=ctx, resilient_parsing=True)
+ args = ctx.protected_args + ctx.args
+ else:
+ sub_ctx = ctx
+
+ while args:
+ name, cmd, args = command.resolve_command(ctx, args)
+
+ if cmd is None:
+ return ctx
+
+ sub_ctx = cmd.make_context(
+ name,
+ args,
+ parent=ctx,
+ allow_extra_args=True,
+ allow_interspersed_args=False,
+ resilient_parsing=True,
+ )
+ args = sub_ctx.args
+
+ ctx = sub_ctx
+ args = [*sub_ctx.protected_args, *sub_ctx.args]
+ else:
+ break
+
+ return ctx
-def _resolve_incomplete(ctx: Context, args: t.List[str], incomplete: str
- ) ->t.Tuple[t.Union[BaseCommand, Parameter], str]:
+def _resolve_incomplete(
+ ctx: Context, args: t.List[str], incomplete: str
+) -> t.Tuple[t.Union[BaseCommand, Parameter], str]:
"""Find the Click object that will handle the completion of the
incomplete value. Return the object and the incomplete value.
@@ -335,4 +560,37 @@ def _resolve_incomplete(ctx: Context, args: t.List[str], incomplete: str
:param args: List of complete args before the incomplete value.
:param incomplete: Value being completed. May be empty.
"""
- pass
+ # Different shells treat an "=" between a long option name and
+ # value differently. Might keep the value joined, return the "="
+ # as a separate item, or return the split name and value. Always
+ # split and discard the "=" to make completion easier.
+ if incomplete == "=":
+ incomplete = ""
+ elif "=" in incomplete and _start_of_option(ctx, incomplete):
+ name, _, incomplete = incomplete.partition("=")
+ args.append(name)
+
+ # The "--" marker tells Click to stop treating values as options
+ # even if they start with the option character. If it hasn't been
+ # given and the incomplete arg looks like an option, the current
+ # command will provide option name completions.
+ if "--" not in args and _start_of_option(ctx, incomplete):
+ return ctx.command, incomplete
+
+ params = ctx.command.get_params(ctx)
+
+ # If the last complete arg is an option name with an incomplete
+ # value, the option will provide value completions.
+ for param in params:
+ if _is_incomplete_option(ctx, args, param):
+ return param, incomplete
+
+ # It's not an option name or value. The first argument without a
+ # parsed value will provide value completions.
+ for param in params:
+ if _is_incomplete_argument(ctx, param):
+ return param, incomplete
+
+ # There were no unparsed arguments, the command may be a group that
+ # will provide command name completions.
+ return ctx.command, incomplete
diff --git a/src/click/termui.py b/src/click/termui.py
index 277721a..db7a4b2 100644
--- a/src/click/termui.py
+++ b/src/click/termui.py
@@ -4,6 +4,7 @@ import itertools
import sys
import typing as t
from gettext import gettext as _
+
from ._compat import isatty
from ._compat import strip_ansi
from .exceptions import Abort
@@ -14,23 +15,79 @@ from .types import convert_type
from .types import ParamType
from .utils import echo
from .utils import LazyFile
+
if t.TYPE_CHECKING:
from ._termui_impl import ProgressBar
-V = t.TypeVar('V')
+
+V = t.TypeVar("V")
+
+# The prompt functions to use. The doc tools currently override these
+# functions to customize how they work.
visible_prompt_func: t.Callable[[str], str] = input
-_ansi_colors = {'black': 30, 'red': 31, 'green': 32, 'yellow': 33, 'blue':
- 34, 'magenta': 35, 'cyan': 36, 'white': 37, 'reset': 39, 'bright_black':
- 90, 'bright_red': 91, 'bright_green': 92, 'bright_yellow': 93,
- 'bright_blue': 94, 'bright_magenta': 95, 'bright_cyan': 96,
- 'bright_white': 97}
-_ansi_reset_all = '\x1b[0m'
-
-
-def prompt(text: str, default: t.Optional[t.Any]=None, hide_input: bool=
- False, confirmation_prompt: t.Union[bool, str]=False, type: t.Optional[
- t.Union[ParamType, t.Any]]=None, value_proc: t.Optional[t.Callable[[str
- ], t.Any]]=None, prompt_suffix: str=': ', show_default: bool=True, err:
- bool=False, show_choices: bool=True) ->t.Any:
+
+_ansi_colors = {
+ "black": 30,
+ "red": 31,
+ "green": 32,
+ "yellow": 33,
+ "blue": 34,
+ "magenta": 35,
+ "cyan": 36,
+ "white": 37,
+ "reset": 39,
+ "bright_black": 90,
+ "bright_red": 91,
+ "bright_green": 92,
+ "bright_yellow": 93,
+ "bright_blue": 94,
+ "bright_magenta": 95,
+ "bright_cyan": 96,
+ "bright_white": 97,
+}
+_ansi_reset_all = "\033[0m"
+
+
+def hidden_prompt_func(prompt: str) -> str:
+ import getpass
+
+ return getpass.getpass(prompt)
+
+
+def _build_prompt(
+ text: str,
+ suffix: str,
+ show_default: bool = False,
+ default: t.Optional[t.Any] = None,
+ show_choices: bool = True,
+ type: t.Optional[ParamType] = None,
+) -> str:
+ prompt = text
+ if type is not None and show_choices and isinstance(type, Choice):
+ prompt += f" ({', '.join(map(str, type.choices))})"
+ if default is not None and show_default:
+ prompt = f"{prompt} [{_format_default(default)}]"
+ return f"{prompt}{suffix}"
+
+
+def _format_default(default: t.Any) -> t.Any:
+ if isinstance(default, (io.IOBase, LazyFile)) and hasattr(default, "name"):
+ return default.name
+
+ return default
+
+
+def prompt(
+ text: str,
+ default: t.Optional[t.Any] = None,
+ hide_input: bool = False,
+ confirmation_prompt: t.Union[bool, str] = False,
+ type: t.Optional[t.Union[ParamType, t.Any]] = None,
+ value_proc: t.Optional[t.Callable[[str], t.Any]] = None,
+ prompt_suffix: str = ": ",
+ show_default: bool = True,
+ err: bool = False,
+ show_choices: bool = True,
+) -> t.Any:
"""Prompts a user for input. This is a convenience function that can
be used to prompt a user for input later.
@@ -71,11 +128,73 @@ def prompt(text: str, default: t.Optional[t.Any]=None, hide_input: bool=
Added the `err` parameter.
"""
- pass
-
-def confirm(text: str, default: t.Optional[bool]=False, abort: bool=False,
- prompt_suffix: str=': ', show_default: bool=True, err: bool=False) ->bool:
+ def prompt_func(text: str) -> str:
+ f = hidden_prompt_func if hide_input else visible_prompt_func
+ try:
+ # Write the prompt separately so that we get nice
+ # coloring through colorama on Windows
+ echo(text.rstrip(" "), nl=False, err=err)
+ # Echo a space to stdout to work around an issue where
+ # readline causes backspace to clear the whole line.
+ return f(" ")
+ except (KeyboardInterrupt, EOFError):
+ # getpass doesn't print a newline if the user aborts input with ^C.
+ # Allegedly this behavior is inherited from getpass(3).
+ # A doc bug has been filed at https://bugs.python.org/issue24711
+ if hide_input:
+ echo(None, err=err)
+ raise Abort() from None
+
+ if value_proc is None:
+ value_proc = convert_type(type, default)
+
+ prompt = _build_prompt(
+ text, prompt_suffix, show_default, default, show_choices, type
+ )
+
+ if confirmation_prompt:
+ if confirmation_prompt is True:
+ confirmation_prompt = _("Repeat for confirmation")
+
+ confirmation_prompt = _build_prompt(confirmation_prompt, prompt_suffix)
+
+ while True:
+ while True:
+ value = prompt_func(prompt)
+ if value:
+ break
+ elif default is not None:
+ value = default
+ break
+ try:
+ result = value_proc(value)
+ except UsageError as e:
+ if hide_input:
+ echo(_("Error: The value you entered was invalid."), err=err)
+ else:
+ echo(_("Error: {e.message}").format(e=e), err=err) # noqa: B306
+ continue
+ if not confirmation_prompt:
+ return result
+ while True:
+ value2 = prompt_func(confirmation_prompt)
+ is_empty = not value and not value2
+ if value2 or is_empty:
+ break
+ if value == value2:
+ return result
+ echo(_("Error: The two entered values do not match."), err=err)
+
+
+def confirm(
+ text: str,
+ default: t.Optional[bool] = False,
+ abort: bool = False,
+ prompt_suffix: str = ": ",
+ show_default: bool = True,
+ err: bool = False,
+) -> bool:
"""Prompts for confirmation (yes/no question).
If the user aborts the input by sending a interrupt signal this
@@ -97,11 +216,42 @@ def confirm(text: str, default: t.Optional[bool]=False, abort: bool=False,
.. versionadded:: 4.0
Added the ``err`` parameter.
"""
- pass
-
-
-def echo_via_pager(text_or_generator: t.Union[t.Iterable[str], t.Callable[[
- ], t.Iterable[str]], str], color: t.Optional[bool]=None) ->None:
+ prompt = _build_prompt(
+ text,
+ prompt_suffix,
+ show_default,
+ "y/n" if default is None else ("Y/n" if default else "y/N"),
+ )
+
+ while True:
+ try:
+ # Write the prompt separately so that we get nice
+ # coloring through colorama on Windows
+ echo(prompt.rstrip(" "), nl=False, err=err)
+ # Echo a space to stdout to work around an issue where
+ # readline causes backspace to clear the whole line.
+ value = visible_prompt_func(" ").lower().strip()
+ except (KeyboardInterrupt, EOFError):
+ raise Abort() from None
+ if value in ("y", "yes"):
+ rv = True
+ elif value in ("n", "no"):
+ rv = False
+ elif default is not None and value == "":
+ rv = default
+ else:
+ echo(_("Error: invalid input"), err=err)
+ continue
+ break
+ if abort and not rv:
+ raise Abort()
+ return rv
+
+
+def echo_via_pager(
+ text_or_generator: t.Union[t.Iterable[str], t.Callable[[], t.Iterable[str]], str],
+ color: t.Optional[bool] = None,
+) -> None:
"""This function takes a text and shows it via an environment specific
pager on stdout.
@@ -113,17 +263,40 @@ def echo_via_pager(text_or_generator: t.Union[t.Iterable[str], t.Callable[[
:param color: controls if the pager supports ANSI colors or not. The
default is autodetection.
"""
- pass
-
-
-def progressbar(iterable: t.Optional[t.Iterable[V]]=None, length: t.
- Optional[int]=None, label: t.Optional[str]=None, show_eta: bool=True,
- show_percent: t.Optional[bool]=None, show_pos: bool=False,
- item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]
- ]=None, fill_char: str='#', empty_char: str='-', bar_template: str=
- '%(label)s [%(bar)s] %(info)s', info_sep: str=' ', width: int=36,
- file: t.Optional[t.TextIO]=None, color: t.Optional[bool]=None,
- update_min_steps: int=1) ->'ProgressBar[V]':
+ color = resolve_color_default(color)
+
+ if inspect.isgeneratorfunction(text_or_generator):
+ i = t.cast(t.Callable[[], t.Iterable[str]], text_or_generator)()
+ elif isinstance(text_or_generator, str):
+ i = [text_or_generator]
+ else:
+ i = iter(t.cast(t.Iterable[str], text_or_generator))
+
+ # convert every element of i to a text type if necessary
+ text_generator = (el if isinstance(el, str) else str(el) for el in i)
+
+ from ._termui_impl import pager
+
+ return pager(itertools.chain(text_generator, "\n"), color)
+
+
+def progressbar(
+ iterable: t.Optional[t.Iterable[V]] = None,
+ length: t.Optional[int] = None,
+ label: t.Optional[str] = None,
+ show_eta: bool = True,
+ show_percent: t.Optional[bool] = None,
+ show_pos: bool = False,
+ item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None,
+ fill_char: str = "#",
+ empty_char: str = "-",
+ bar_template: str = "%(label)s [%(bar)s] %(info)s",
+ info_sep: str = " ",
+ width: int = 36,
+ file: t.Optional[t.TextIO] = None,
+ color: t.Optional[bool] = None,
+ update_min_steps: int = 1,
+) -> "ProgressBar[V]":
"""This function creates an iterable context manager that can be used
to iterate over something while showing a progress bar. It will
either iterate over the `iterable` or `length` items (that are counted
@@ -237,26 +410,69 @@ def progressbar(iterable: t.Optional[t.Iterable[V]]=None, length: t.
.. versionadded:: 2.0
"""
- pass
-
+ from ._termui_impl import ProgressBar
-def clear() ->None:
+ color = resolve_color_default(color)
+ return ProgressBar(
+ iterable=iterable,
+ length=length,
+ show_eta=show_eta,
+ show_percent=show_percent,
+ show_pos=show_pos,
+ item_show_func=item_show_func,
+ fill_char=fill_char,
+ empty_char=empty_char,
+ bar_template=bar_template,
+ info_sep=info_sep,
+ file=file,
+ label=label,
+ width=width,
+ color=color,
+ update_min_steps=update_min_steps,
+ )
+
+
+def clear() -> None:
"""Clears the terminal screen. This will have the effect of clearing
the whole visible space of the terminal and moving the cursor to the
top left. This does not do anything if not connected to a terminal.
.. versionadded:: 2.0
"""
- pass
-
-
-def style(text: t.Any, fg: t.Optional[t.Union[int, t.Tuple[int, int, int],
- str]]=None, bg: t.Optional[t.Union[int, t.Tuple[int, int, int], str]]=
- None, bold: t.Optional[bool]=None, dim: t.Optional[bool]=None,
- underline: t.Optional[bool]=None, overline: t.Optional[bool]=None,
- italic: t.Optional[bool]=None, blink: t.Optional[bool]=None, reverse: t
- .Optional[bool]=None, strikethrough: t.Optional[bool]=None, reset: bool
- =True) ->str:
+ if not isatty(sys.stdout):
+ return
+
+ # ANSI escape \033[2J clears the screen, \033[1;1H moves the cursor
+ echo("\033[2J\033[1;1H", nl=False)
+
+
+def _interpret_color(
+ color: t.Union[int, t.Tuple[int, int, int], str], offset: int = 0
+) -> str:
+ if isinstance(color, int):
+ return f"{38 + offset};5;{color:d}"
+
+ if isinstance(color, (tuple, list)):
+ r, g, b = color
+ return f"{38 + offset};2;{r:d};{g:d};{b:d}"
+
+ return str(_ansi_colors[color] + offset)
+
+
+def style(
+ text: t.Any,
+ fg: t.Optional[t.Union[int, t.Tuple[int, int, int], str]] = None,
+ bg: t.Optional[t.Union[int, t.Tuple[int, int, int], str]] = None,
+ bold: t.Optional[bool] = None,
+ dim: t.Optional[bool] = None,
+ underline: t.Optional[bool] = None,
+ overline: t.Optional[bool] = None,
+ italic: t.Optional[bool] = None,
+ blink: t.Optional[bool] = None,
+ reverse: t.Optional[bool] = None,
+ strikethrough: t.Optional[bool] = None,
+ reset: bool = True,
+) -> str:
"""Styles a text with ANSI styles and returns the new string. By
default the styling is self contained which means that at the end
of the string a reset code is issued. This can be prevented by
@@ -333,10 +549,46 @@ def style(text: t.Any, fg: t.Optional[t.Union[int, t.Tuple[int, int, int],
.. versionadded:: 2.0
"""
- pass
-
-
-def unstyle(text: str) ->str:
+ if not isinstance(text, str):
+ text = str(text)
+
+ bits = []
+
+ if fg:
+ try:
+ bits.append(f"\033[{_interpret_color(fg)}m")
+ except KeyError:
+ raise TypeError(f"Unknown color {fg!r}") from None
+
+ if bg:
+ try:
+ bits.append(f"\033[{_interpret_color(bg, 10)}m")
+ except KeyError:
+ raise TypeError(f"Unknown color {bg!r}") from None
+
+ if bold is not None:
+ bits.append(f"\033[{1 if bold else 22}m")
+ if dim is not None:
+ bits.append(f"\033[{2 if dim else 22}m")
+ if underline is not None:
+ bits.append(f"\033[{4 if underline else 24}m")
+ if overline is not None:
+ bits.append(f"\033[{53 if overline else 55}m")
+ if italic is not None:
+ bits.append(f"\033[{3 if italic else 23}m")
+ if blink is not None:
+ bits.append(f"\033[{5 if blink else 25}m")
+ if reverse is not None:
+ bits.append(f"\033[{7 if reverse else 27}m")
+ if strikethrough is not None:
+ bits.append(f"\033[{9 if strikethrough else 29}m")
+ bits.append(text)
+ if reset:
+ bits.append(_ansi_reset_all)
+ return "".join(bits)
+
+
+def unstyle(text: str) -> str:
"""Removes ANSI styling information from a string. Usually it's not
necessary to use this function as Click's echo function will
automatically remove styling if necessary.
@@ -345,12 +597,17 @@ def unstyle(text: str) ->str:
:param text: the text to remove style information from.
"""
- pass
+ return strip_ansi(text)
-def secho(message: t.Optional[t.Any]=None, file: t.Optional[t.IO[t.AnyStr]]
- =None, nl: bool=True, err: bool=False, color: t.Optional[bool]=None, **
- styles: t.Any) ->None:
+def secho(
+ message: t.Optional[t.Any] = None,
+ file: t.Optional[t.IO[t.AnyStr]] = None,
+ nl: bool = True,
+ err: bool = False,
+ color: t.Optional[bool] = None,
+ **styles: t.Any,
+) -> None:
"""This function combines :func:`echo` and :func:`style` into one
call. As such the following two calls are the same::
@@ -371,14 +628,21 @@ def secho(message: t.Optional[t.Any]=None, file: t.Optional[t.IO[t.AnyStr]]
.. versionadded:: 2.0
"""
- pass
+ if message is not None and not isinstance(message, (bytes, bytearray)):
+ message = style(message, **styles)
+ return echo(message, file=file, nl=nl, err=err, color=color)
-def edit(text: t.Optional[t.AnyStr]=None, editor: t.Optional[str]=None, env:
- t.Optional[t.Mapping[str, str]]=None, require_save: bool=True,
- extension: str='.txt', filename: t.Optional[str]=None) ->t.Optional[t.
- AnyStr]:
- """Edits the given text in the defined editor. If an editor is given
+
+def edit(
+ text: t.Optional[t.AnyStr] = None,
+ editor: t.Optional[str] = None,
+ env: t.Optional[t.Mapping[str, str]] = None,
+ require_save: bool = True,
+ extension: str = ".txt",
+ filename: t.Optional[str] = None,
+) -> t.Optional[t.AnyStr]:
+ r"""Edits the given text in the defined editor. If an editor is given
(should be the full path to the executable but the regular operating
system search path is used for finding the executable) it overrides
the detected editor. Optionally, some environment variables can be
@@ -390,7 +654,7 @@ def edit(text: t.Optional[t.AnyStr]=None, editor: t.Optional[str]=None, env:
Note for Windows: to simplify cross-platform usage, the newlines are
automatically converted from POSIX to Windows and vice versa. As such,
- the message here will have ``\\n`` as newline markers.
+ the message here will have ``\n`` as newline markers.
:param text: the text to edit.
:param editor: optionally the editor to use. Defaults to automatic
@@ -405,10 +669,18 @@ def edit(text: t.Optional[t.AnyStr]=None, editor: t.Optional[str]=None, env:
provided text contents. It will not use a temporary
file as an indirection in that case.
"""
- pass
+ from ._termui_impl import Editor
+
+ ed = Editor(editor=editor, env=env, require_save=require_save, extension=extension)
+ if filename is None:
+ return ed.edit(text)
-def launch(url: str, wait: bool=False, locate: bool=False) ->int:
+ ed.edit_file(filename)
+ return None
+
+
+def launch(url: str, wait: bool = False, locate: bool = False) -> int:
"""This function launches the given URL (or filename) in the default
viewer application for this file type. If this is an executable, it
might launch the executable in a new session. The return value is
@@ -432,13 +704,17 @@ def launch(url: str, wait: bool=False, locate: bool=False) ->int:
might have weird effects if the URL does not point to
the filesystem.
"""
- pass
+ from ._termui_impl import open_url
+
+ return open_url(url, wait=wait, locate=locate)
+# If this is provided, getchar() calls into this instead. This is used
+# for unittesting purposes.
_getchar: t.Optional[t.Callable[[bool], str]] = None
-def getchar(echo: bool=False) ->str:
+def getchar(echo: bool = False) -> str:
"""Fetches a single character from the terminal and returns it. This
will always return a unicode character and under certain rare
circumstances this might return more than one character. The
@@ -458,10 +734,23 @@ def getchar(echo: bool=False) ->str:
:param echo: if set to `True`, the character read will also show up on
the terminal. The default is to not show it.
"""
- pass
+ global _getchar
+
+ if _getchar is None:
+ from ._termui_impl import getchar as f
+
+ _getchar = f
+
+ return _getchar(echo)
+
+
+def raw_terminal() -> t.ContextManager[int]:
+ from ._termui_impl import raw_terminal as f
+
+ return f()
-def pause(info: t.Optional[str]=None, err: bool=False) ->None:
+def pause(info: t.Optional[str] = None, err: bool = False) -> None:
"""This command stops execution and waits for the user to press any
key to continue. This is similar to the Windows batch "pause"
command. If the program is not run through a terminal, this command
@@ -477,4 +766,19 @@ def pause(info: t.Optional[str]=None, err: bool=False) ->None:
:param err: if set to message goes to ``stderr`` instead of
``stdout``, the same as with echo.
"""
- pass
+ if not isatty(sys.stdin) or not isatty(sys.stdout):
+ return
+
+ if info is None:
+ info = _("Press any key to continue...")
+
+ try:
+ if info:
+ echo(info, nl=False, err=err)
+ try:
+ getchar()
+ except (KeyboardInterrupt, EOFError):
+ pass
+ finally:
+ if info:
+ echo(err=err)
diff --git a/src/click/testing.py b/src/click/testing.py
index 320d223..e0df0d2 100644
--- a/src/click/testing.py
+++ b/src/click/testing.py
@@ -7,73 +7,153 @@ import sys
import tempfile
import typing as t
from types import TracebackType
+
from . import formatting
from . import termui
from . import utils
from ._compat import _find_binary_reader
+
if t.TYPE_CHECKING:
from .core import BaseCommand
class EchoingStdin:
-
- def __init__(self, input: t.BinaryIO, output: t.BinaryIO) ->None:
+ def __init__(self, input: t.BinaryIO, output: t.BinaryIO) -> None:
self._input = input
self._output = output
self._paused = False
- def __getattr__(self, x: str) ->t.Any:
+ def __getattr__(self, x: str) -> t.Any:
return getattr(self._input, x)
- def __iter__(self) ->t.Iterator[bytes]:
+ def _echo(self, rv: bytes) -> bytes:
+ if not self._paused:
+ self._output.write(rv)
+
+ return rv
+
+ def read(self, n: int = -1) -> bytes:
+ return self._echo(self._input.read(n))
+
+ def read1(self, n: int = -1) -> bytes:
+ return self._echo(self._input.read1(n)) # type: ignore
+
+ def readline(self, n: int = -1) -> bytes:
+ return self._echo(self._input.readline(n))
+
+ def readlines(self) -> t.List[bytes]:
+ return [self._echo(x) for x in self._input.readlines()]
+
+ def __iter__(self) -> t.Iterator[bytes]:
return iter(self._echo(x) for x in self._input)
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
return repr(self._input)
-class _NamedTextIOWrapper(io.TextIOWrapper):
+@contextlib.contextmanager
+def _pause_echo(stream: t.Optional[EchoingStdin]) -> t.Iterator[None]:
+ if stream is None:
+ yield
+ else:
+ stream._paused = True
+ yield
+ stream._paused = False
+
- def __init__(self, buffer: t.BinaryIO, name: str, mode: str, **kwargs:
- t.Any) ->None:
+class _NamedTextIOWrapper(io.TextIOWrapper):
+ def __init__(
+ self, buffer: t.BinaryIO, name: str, mode: str, **kwargs: t.Any
+ ) -> None:
super().__init__(buffer, **kwargs)
self._name = name
self._mode = mode
+ @property
+ def name(self) -> str:
+ return self._name
+
+ @property
+ def mode(self) -> str:
+ return self._mode
+
+
+def make_input_stream(
+ input: t.Optional[t.Union[str, bytes, t.IO[t.Any]]], charset: str
+) -> t.BinaryIO:
+ # Is already an input stream.
+ if hasattr(input, "read"):
+ rv = _find_binary_reader(t.cast(t.IO[t.Any], input))
+
+ if rv is not None:
+ return rv
+
+ raise TypeError("Could not find binary reader for input stream.")
+
+ if input is None:
+ input = b""
+ elif isinstance(input, str):
+ input = input.encode(charset)
+
+ return io.BytesIO(input)
+
class Result:
"""Holds the captured result of an invoked CLI script."""
- def __init__(self, runner: 'CliRunner', stdout_bytes: bytes,
- stderr_bytes: t.Optional[bytes], return_value: t.Any, exit_code:
- int, exception: t.Optional[BaseException], exc_info: t.Optional[t.
- Tuple[t.Type[BaseException], BaseException, TracebackType]]=None):
+ def __init__(
+ self,
+ runner: "CliRunner",
+ stdout_bytes: bytes,
+ stderr_bytes: t.Optional[bytes],
+ return_value: t.Any,
+ exit_code: int,
+ exception: t.Optional[BaseException],
+ exc_info: t.Optional[
+ t.Tuple[t.Type[BaseException], BaseException, TracebackType]
+ ] = None,
+ ):
+ #: The runner that created the result
self.runner = runner
+ #: The standard output as bytes.
self.stdout_bytes = stdout_bytes
+ #: The standard error as bytes, or None if not available
self.stderr_bytes = stderr_bytes
+ #: The value returned from the invoked command.
+ #:
+ #: .. versionadded:: 8.0
self.return_value = return_value
+ #: The exit code as integer.
self.exit_code = exit_code
+ #: The exception that happened if one did.
self.exception = exception
+ #: The traceback
self.exc_info = exc_info
@property
- def output(self) ->str:
+ def output(self) -> str:
"""The (standard) output as unicode string."""
- pass
+ return self.stdout
@property
- def stdout(self) ->str:
+ def stdout(self) -> str:
"""The standard output as unicode string."""
- pass
+ return self.stdout_bytes.decode(self.runner.charset, "replace").replace(
+ "\r\n", "\n"
+ )
@property
- def stderr(self) ->str:
+ def stderr(self) -> str:
"""The standard error as unicode string."""
- pass
+ if self.stderr_bytes is None:
+ raise ValueError("stderr not separately captured")
+ return self.stderr_bytes.decode(self.runner.charset, "replace").replace(
+ "\r\n", "\n"
+ )
- def __repr__(self) ->str:
- exc_str = repr(self.exception) if self.exception else 'okay'
- return f'<{type(self).__name__} {exc_str}>'
+ def __repr__(self) -> str:
+ exc_str = repr(self.exception) if self.exception else "okay"
+ return f"<{type(self).__name__} {exc_str}>"
class CliRunner:
@@ -95,30 +175,41 @@ class CliRunner:
independently
"""
- def __init__(self, charset: str='utf-8', env: t.Optional[t.Mapping[str,
- t.Optional[str]]]=None, echo_stdin: bool=False, mix_stderr: bool=True
- ) ->None:
+ def __init__(
+ self,
+ charset: str = "utf-8",
+ env: t.Optional[t.Mapping[str, t.Optional[str]]] = None,
+ echo_stdin: bool = False,
+ mix_stderr: bool = True,
+ ) -> None:
self.charset = charset
self.env: t.Mapping[str, t.Optional[str]] = env or {}
self.echo_stdin = echo_stdin
self.mix_stderr = mix_stderr
- def get_default_prog_name(self, cli: 'BaseCommand') ->str:
+ def get_default_prog_name(self, cli: "BaseCommand") -> str:
"""Given a command object it will return the default program name
for it. The default is the `name` attribute or ``"root"`` if not
set.
"""
- pass
+ return cli.name or "root"
- def make_env(self, overrides: t.Optional[t.Mapping[str, t.Optional[str]
- ]]=None) ->t.Mapping[str, t.Optional[str]]:
+ def make_env(
+ self, overrides: t.Optional[t.Mapping[str, t.Optional[str]]] = None
+ ) -> t.Mapping[str, t.Optional[str]]:
"""Returns the environment overrides for invoking a script."""
- pass
+ rv = dict(self.env)
+ if overrides:
+ rv.update(overrides)
+ return rv
@contextlib.contextmanager
- def isolation(self, input: t.Optional[t.Union[str, bytes, t.IO[t.Any]]]
- =None, env: t.Optional[t.Mapping[str, t.Optional[str]]]=None, color:
- bool=False) ->t.Iterator[t.Tuple[io.BytesIO, t.Optional[io.BytesIO]]]:
+ def isolation(
+ self,
+ input: t.Optional[t.Union[str, bytes, t.IO[t.Any]]] = None,
+ env: t.Optional[t.Mapping[str, t.Optional[str]]] = None,
+ color: bool = False,
+ ) -> t.Iterator[t.Tuple[io.BytesIO, t.Optional[io.BytesIO]]]:
"""A context manager that sets up the isolation for invoking of a
command line tool. This sets up stdin with the given input data
and `os.environ` with the overrides from the given dictionary.
@@ -139,13 +230,132 @@ class CliRunner:
.. versionchanged:: 4.0
Added the ``color`` parameter.
"""
- pass
-
- def invoke(self, cli: 'BaseCommand', args: t.Optional[t.Union[str, t.
- Sequence[str]]]=None, input: t.Optional[t.Union[str, bytes, t.IO[t.
- Any]]]=None, env: t.Optional[t.Mapping[str, t.Optional[str]]]=None,
- catch_exceptions: bool=True, color: bool=False, **extra: t.Any
- ) ->Result:
+ bytes_input = make_input_stream(input, self.charset)
+ echo_input = None
+
+ old_stdin = sys.stdin
+ old_stdout = sys.stdout
+ old_stderr = sys.stderr
+ old_forced_width = formatting.FORCED_WIDTH
+ formatting.FORCED_WIDTH = 80
+
+ env = self.make_env(env)
+
+ bytes_output = io.BytesIO()
+
+ if self.echo_stdin:
+ bytes_input = echo_input = t.cast(
+ t.BinaryIO, EchoingStdin(bytes_input, bytes_output)
+ )
+
+ sys.stdin = text_input = _NamedTextIOWrapper(
+ bytes_input, encoding=self.charset, name="<stdin>", mode="r"
+ )
+
+ if self.echo_stdin:
+ # Force unbuffered reads, otherwise TextIOWrapper reads a
+ # large chunk which is echoed early.
+ text_input._CHUNK_SIZE = 1 # type: ignore
+
+ sys.stdout = _NamedTextIOWrapper(
+ bytes_output, encoding=self.charset, name="<stdout>", mode="w"
+ )
+
+ bytes_error = None
+ if self.mix_stderr:
+ sys.stderr = sys.stdout
+ else:
+ bytes_error = io.BytesIO()
+ sys.stderr = _NamedTextIOWrapper(
+ bytes_error,
+ encoding=self.charset,
+ name="<stderr>",
+ mode="w",
+ errors="backslashreplace",
+ )
+
+ @_pause_echo(echo_input) # type: ignore
+ def visible_input(prompt: t.Optional[str] = None) -> str:
+ sys.stdout.write(prompt or "")
+ val = text_input.readline().rstrip("\r\n")
+ sys.stdout.write(f"{val}\n")
+ sys.stdout.flush()
+ return val
+
+ @_pause_echo(echo_input) # type: ignore
+ def hidden_input(prompt: t.Optional[str] = None) -> str:
+ sys.stdout.write(f"{prompt or ''}\n")
+ sys.stdout.flush()
+ return text_input.readline().rstrip("\r\n")
+
+ @_pause_echo(echo_input) # type: ignore
+ def _getchar(echo: bool) -> str:
+ char = sys.stdin.read(1)
+
+ if echo:
+ sys.stdout.write(char)
+
+ sys.stdout.flush()
+ return char
+
+ default_color = color
+
+ def should_strip_ansi(
+ stream: t.Optional[t.IO[t.Any]] = None, color: t.Optional[bool] = None
+ ) -> bool:
+ if color is None:
+ return not default_color
+ return not color
+
+ old_visible_prompt_func = termui.visible_prompt_func
+ old_hidden_prompt_func = termui.hidden_prompt_func
+ old__getchar_func = termui._getchar
+ old_should_strip_ansi = utils.should_strip_ansi # type: ignore
+ termui.visible_prompt_func = visible_input
+ termui.hidden_prompt_func = hidden_input
+ termui._getchar = _getchar
+ utils.should_strip_ansi = should_strip_ansi # type: ignore
+
+ old_env = {}
+ try:
+ for key, value in env.items():
+ old_env[key] = os.environ.get(key)
+ if value is None:
+ try:
+ del os.environ[key]
+ except Exception:
+ pass
+ else:
+ os.environ[key] = value
+ yield (bytes_output, bytes_error)
+ finally:
+ for key, value in old_env.items():
+ if value is None:
+ try:
+ del os.environ[key]
+ except Exception:
+ pass
+ else:
+ os.environ[key] = value
+ sys.stdout = old_stdout
+ sys.stderr = old_stderr
+ sys.stdin = old_stdin
+ termui.visible_prompt_func = old_visible_prompt_func
+ termui.hidden_prompt_func = old_hidden_prompt_func
+ termui._getchar = old__getchar_func
+ utils.should_strip_ansi = old_should_strip_ansi # type: ignore
+ formatting.FORCED_WIDTH = old_forced_width
+
+ def invoke(
+ self,
+ cli: "BaseCommand",
+ args: t.Optional[t.Union[str, t.Sequence[str]]] = None,
+ input: t.Optional[t.Union[str, bytes, t.IO[t.Any]]] = None,
+ env: t.Optional[t.Mapping[str, t.Optional[str]]] = None,
+ catch_exceptions: bool = True,
+ color: bool = False,
+ **extra: t.Any,
+ ) -> Result:
"""Invokes a command in an isolated environment. The arguments are
forwarded directly to the command line script, the `extra` keyword
arguments are passed to the :meth:`~clickpkg.Command.main` function of
@@ -180,11 +390,67 @@ class CliRunner:
The result object has the ``exc_info`` attribute with the
traceback if available.
"""
- pass
+ exc_info = None
+ with self.isolation(input=input, env=env, color=color) as outstreams:
+ return_value = None
+ exception: t.Optional[BaseException] = None
+ exit_code = 0
+
+ if isinstance(args, str):
+ args = shlex.split(args)
+
+ try:
+ prog_name = extra.pop("prog_name")
+ except KeyError:
+ prog_name = self.get_default_prog_name(cli)
+
+ try:
+ return_value = cli.main(args=args or (), prog_name=prog_name, **extra)
+ except SystemExit as e:
+ exc_info = sys.exc_info()
+ e_code = t.cast(t.Optional[t.Union[int, t.Any]], e.code)
+
+ if e_code is None:
+ e_code = 0
+
+ if e_code != 0:
+ exception = e
+
+ if not isinstance(e_code, int):
+ sys.stdout.write(str(e_code))
+ sys.stdout.write("\n")
+ e_code = 1
+
+ exit_code = e_code
+
+ except Exception as e:
+ if not catch_exceptions:
+ raise
+ exception = e
+ exit_code = 1
+ exc_info = sys.exc_info()
+ finally:
+ sys.stdout.flush()
+ stdout = outstreams[0].getvalue()
+ if self.mix_stderr:
+ stderr = None
+ else:
+ stderr = outstreams[1].getvalue() # type: ignore
+
+ return Result(
+ runner=self,
+ stdout_bytes=stdout,
+ stderr_bytes=stderr,
+ return_value=return_value,
+ exit_code=exit_code,
+ exception=exception,
+ exc_info=exc_info, # type: ignore
+ )
@contextlib.contextmanager
- def isolated_filesystem(self, temp_dir: t.Optional[t.Union[str,
- 'os.PathLike[str]']]=None) ->t.Iterator[str]:
+ def isolated_filesystem(
+ self, temp_dir: t.Optional[t.Union[str, "os.PathLike[str]"]] = None
+ ) -> t.Iterator[str]:
"""A context manager that creates a temporary directory and
changes the current working directory to it. This isolates tests
that affect the contents of the CWD to prevent them from
@@ -197,4 +463,17 @@ class CliRunner:
.. versionchanged:: 8.0
Added the ``temp_dir`` parameter.
"""
- pass
+ cwd = os.getcwd()
+ dt = tempfile.mkdtemp(dir=temp_dir)
+ os.chdir(dt)
+
+ try:
+ yield dt
+ finally:
+ os.chdir(cwd)
+
+ if temp_dir is None:
+ try:
+ shutil.rmtree(dt)
+ except OSError: # noqa: B014
+ pass
diff --git a/src/click/types.py b/src/click/types.py
index 4527388..2b1d179 100644
--- a/src/click/types.py
+++ b/src/click/types.py
@@ -5,12 +5,14 @@ import typing as t
from datetime import datetime
from gettext import gettext as _
from gettext import ngettext
+
from ._compat import _get_argv_encoding
from ._compat import open_stream
from .exceptions import BadParameter
from .utils import format_filename
from .utils import LazyFile
from .utils import safecall
+
if t.TYPE_CHECKING:
import typing_extensions as te
from .core import Context
@@ -35,12 +37,22 @@ class ParamType:
arguments are ``None``. This can occur when converting prompt
input.
"""
+
is_composite: t.ClassVar[bool] = False
arity: t.ClassVar[int] = 1
+
+ #: the descriptive name of this type
name: str
+
+ #: if a list of this type is expected and the value is pulled from a
+ #: string environment variable, this is what splits it up. `None`
+ #: means any whitespace. For all parameters the general rule is that
+ #: whitespace splits them up. The exception are paths and files which
+ #: are split by ``os.path.pathsep`` by default (":" on Unix and ";" on
+ #: Windows).
envvar_list_splitter: t.ClassVar[t.Optional[str]] = None
- def to_info_dict(self) ->t.Dict[str, t.Any]:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
"""Gather information that could be useful for a tool generating
user-facing documentation.
@@ -49,27 +61,40 @@ class ParamType:
.. versionadded:: 8.0
"""
- pass
+ # The class name without the "ParamType" suffix.
+ param_type = type(self).__name__.partition("ParamType")[0]
+ param_type = param_type.partition("ParameterType")[0]
- def __call__(self, value: t.Any, param: t.Optional['Parameter']=None,
- ctx: t.Optional['Context']=None) ->t.Any:
+ # Custom subclasses might not remember to set a name.
+ if hasattr(self, "name"):
+ name = self.name
+ else:
+ name = param_type
+
+ return {"param_type": param_type, "name": name}
+
+ def __call__(
+ self,
+ value: t.Any,
+ param: t.Optional["Parameter"] = None,
+ ctx: t.Optional["Context"] = None,
+ ) -> t.Any:
if value is not None:
return self.convert(value, param, ctx)
- def get_metavar(self, param: 'Parameter') ->t.Optional[str]:
+ def get_metavar(self, param: "Parameter") -> t.Optional[str]:
"""Returns the metavar default for this param if it provides one."""
- pass
- def get_missing_message(self, param: 'Parameter') ->t.Optional[str]:
+ def get_missing_message(self, param: "Parameter") -> t.Optional[str]:
"""Optionally might return extra information about a missing
parameter.
.. versionadded:: 2.0
"""
- pass
- def convert(self, value: t.Any, param: t.Optional['Parameter'], ctx: t.
- Optional['Context']) ->t.Any:
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
"""Convert the value to the correct type. This is not called if
the value is ``None`` (the missing value).
@@ -89,9 +114,9 @@ class ParamType:
:param ctx: The current context that arrived at this value. May
be ``None``.
"""
- pass
+ return value
- def split_envvar_value(self, rv: str) ->t.Sequence[str]:
+ def split_envvar_value(self, rv: str) -> t.Sequence[str]:
"""Given a value from an environment variable this splits it up
into small chunks depending on the defined envvar list splitter.
@@ -99,15 +124,20 @@ class ParamType:
then leading and trailing whitespace is ignored. Otherwise, leading
and trailing splitters usually lead to empty items being included.
"""
- pass
-
- def fail(self, message: str, param: t.Optional['Parameter']=None, ctx:
- t.Optional['Context']=None) ->'t.NoReturn':
+ return (rv or "").split(self.envvar_list_splitter)
+
+ def fail(
+ self,
+ message: str,
+ param: t.Optional["Parameter"] = None,
+ ctx: t.Optional["Context"] = None,
+ ) -> "t.NoReturn":
"""Helper method to fail with an invalid value message."""
- pass
+ raise BadParameter(message, ctx=ctx, param=param)
- def shell_complete(self, ctx: 'Context', param: 'Parameter', incomplete:
- str) ->t.List['CompletionItem']:
+ def shell_complete(
+ self, ctx: "Context", param: "Parameter", incomplete: str
+ ) -> t.List["CompletionItem"]:
"""Return a list of
:class:`~click.shell_completion.CompletionItem` objects for the
incomplete value. Most types do not provide completions, but
@@ -120,32 +150,77 @@ class ParamType:
.. versionadded:: 8.0
"""
- pass
+ return []
class CompositeParamType(ParamType):
is_composite = True
+ @property
+ def arity(self) -> int: # type: ignore
+ raise NotImplementedError()
-class FuncParamType(ParamType):
- def __init__(self, func: t.Callable[[t.Any], t.Any]) ->None:
+class FuncParamType(ParamType):
+ def __init__(self, func: t.Callable[[t.Any], t.Any]) -> None:
self.name: str = func.__name__
self.func = func
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict["func"] = self.func
+ return info_dict
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ try:
+ return self.func(value)
+ except ValueError:
+ try:
+ value = str(value)
+ except UnicodeError:
+ value = value.decode("utf-8", "replace")
+
+ self.fail(value, param, ctx)
+
class UnprocessedParamType(ParamType):
- name = 'text'
+ name = "text"
- def __repr__(self) ->str:
- return 'UNPROCESSED'
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ return value
+ def __repr__(self) -> str:
+ return "UNPROCESSED"
-class StringParamType(ParamType):
- name = 'text'
- def __repr__(self) ->str:
- return 'STRING'
+class StringParamType(ParamType):
+ name = "text"
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ if isinstance(value, bytes):
+ enc = _get_argv_encoding()
+ try:
+ value = value.decode(enc)
+ except UnicodeError:
+ fs_enc = sys.getfilesystemencoding()
+ if fs_enc != enc:
+ try:
+ value = value.decode(fs_enc)
+ except UnicodeError:
+ value = value.decode("utf-8", "replace")
+ else:
+ value = value.decode("utf-8", "replace")
+ return value
+ return str(value)
+
+ def __repr__(self) -> str:
+ return "STRING"
class Choice(ParamType):
@@ -164,18 +239,76 @@ class Choice(ParamType):
:param case_sensitive: Set to false to make choices case
insensitive. Defaults to true.
"""
- name = 'choice'
- def __init__(self, choices: t.Sequence[str], case_sensitive: bool=True
- ) ->None:
+ name = "choice"
+
+ def __init__(self, choices: t.Sequence[str], case_sensitive: bool = True) -> None:
self.choices = choices
self.case_sensitive = case_sensitive
- def __repr__(self) ->str:
- return f'Choice({list(self.choices)})'
-
- def shell_complete(self, ctx: 'Context', param: 'Parameter', incomplete:
- str) ->t.List['CompletionItem']:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict["choices"] = self.choices
+ info_dict["case_sensitive"] = self.case_sensitive
+ return info_dict
+
+ def get_metavar(self, param: "Parameter") -> str:
+ choices_str = "|".join(self.choices)
+
+ # Use curly braces to indicate a required argument.
+ if param.required and param.param_type_name == "argument":
+ return f"{{{choices_str}}}"
+
+ # Use square braces to indicate an option or optional argument.
+ return f"[{choices_str}]"
+
+ def get_missing_message(self, param: "Parameter") -> str:
+ return _("Choose from:\n\t{choices}").format(choices=",\n\t".join(self.choices))
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ # Match through normalization and case sensitivity
+ # first do token_normalize_func, then lowercase
+ # preserve original `value` to produce an accurate message in
+ # `self.fail`
+ normed_value = value
+ normed_choices = {choice: choice for choice in self.choices}
+
+ if ctx is not None and ctx.token_normalize_func is not None:
+ normed_value = ctx.token_normalize_func(value)
+ normed_choices = {
+ ctx.token_normalize_func(normed_choice): original
+ for normed_choice, original in normed_choices.items()
+ }
+
+ if not self.case_sensitive:
+ normed_value = normed_value.casefold()
+ normed_choices = {
+ normed_choice.casefold(): original
+ for normed_choice, original in normed_choices.items()
+ }
+
+ if normed_value in normed_choices:
+ return normed_choices[normed_value]
+
+ choices_str = ", ".join(map(repr, self.choices))
+ self.fail(
+ ngettext(
+ "{value!r} is not {choice}.",
+ "{value!r} is not one of {choices}.",
+ len(self.choices),
+ ).format(value=value, choice=choices_str, choices=choices_str),
+ param,
+ ctx,
+ )
+
+ def __repr__(self) -> str:
+ return f"Choice({list(self.choices)})"
+
+ def shell_complete(
+ self, ctx: "Context", param: "Parameter", incomplete: str
+ ) -> t.List["CompletionItem"]:
"""Complete choices that start with the incomplete value.
:param ctx: Invocation context for this command.
@@ -184,7 +317,17 @@ class Choice(ParamType):
.. versionadded:: 8.0
"""
- pass
+ from click.shell_completion import CompletionItem
+
+ str_choices = map(str, self.choices)
+
+ if self.case_sensitive:
+ matched = (c for c in str_choices if c.startswith(incomplete))
+ else:
+ incomplete = incomplete.lower()
+ matched = (c for c in str_choices if c.lower().startswith(incomplete))
+
+ return [CompletionItem(c) for c in matched]
class DateTime(ParamType):
@@ -207,33 +350,133 @@ class DateTime(ParamType):
``'%Y-%m-%d'``, ``'%Y-%m-%dT%H:%M:%S'``,
``'%Y-%m-%d %H:%M:%S'``.
"""
- name = 'datetime'
- def __init__(self, formats: t.Optional[t.Sequence[str]]=None):
- self.formats: t.Sequence[str] = formats or ['%Y-%m-%d',
- '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S']
+ name = "datetime"
+
+ def __init__(self, formats: t.Optional[t.Sequence[str]] = None):
+ self.formats: t.Sequence[str] = formats or [
+ "%Y-%m-%d",
+ "%Y-%m-%dT%H:%M:%S",
+ "%Y-%m-%d %H:%M:%S",
+ ]
+
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict["formats"] = self.formats
+ return info_dict
+
+ def get_metavar(self, param: "Parameter") -> str:
+ return f"[{'|'.join(self.formats)}]"
+
+ def _try_to_convert_date(self, value: t.Any, format: str) -> t.Optional[datetime]:
+ try:
+ return datetime.strptime(value, format)
+ except ValueError:
+ return None
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ if isinstance(value, datetime):
+ return value
+
+ for format in self.formats:
+ converted = self._try_to_convert_date(value, format)
+
+ if converted is not None:
+ return converted
+
+ formats_str = ", ".join(map(repr, self.formats))
+ self.fail(
+ ngettext(
+ "{value!r} does not match the format {format}.",
+ "{value!r} does not match the formats {formats}.",
+ len(self.formats),
+ ).format(value=value, format=formats_str, formats=formats_str),
+ param,
+ ctx,
+ )
- def __repr__(self) ->str:
- return 'DateTime'
+ def __repr__(self) -> str:
+ return "DateTime"
class _NumberParamTypeBase(ParamType):
_number_class: t.ClassVar[t.Type[t.Any]]
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ try:
+ return self._number_class(value)
+ except ValueError:
+ self.fail(
+ _("{value!r} is not a valid {number_type}.").format(
+ value=value, number_type=self.name
+ ),
+ param,
+ ctx,
+ )
-class _NumberRangeBase(_NumberParamTypeBase):
- def __init__(self, min: t.Optional[float]=None, max: t.Optional[float]=
- None, min_open: bool=False, max_open: bool=False, clamp: bool=False
- ) ->None:
+class _NumberRangeBase(_NumberParamTypeBase):
+ def __init__(
+ self,
+ min: t.Optional[float] = None,
+ max: t.Optional[float] = None,
+ min_open: bool = False,
+ max_open: bool = False,
+ clamp: bool = False,
+ ) -> None:
self.min = min
self.max = max
self.min_open = min_open
self.max_open = max_open
self.clamp = clamp
- def _clamp(self, bound: float, dir: 'te.Literal[1, -1]', open: bool
- ) ->float:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict.update(
+ min=self.min,
+ max=self.max,
+ min_open=self.min_open,
+ max_open=self.max_open,
+ clamp=self.clamp,
+ )
+ return info_dict
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ import operator
+
+ rv = super().convert(value, param, ctx)
+ lt_min: bool = self.min is not None and (
+ operator.le if self.min_open else operator.lt
+ )(rv, self.min)
+ gt_max: bool = self.max is not None and (
+ operator.ge if self.max_open else operator.gt
+ )(rv, self.max)
+
+ if self.clamp:
+ if lt_min:
+ return self._clamp(self.min, 1, self.min_open) # type: ignore
+
+ if gt_max:
+ return self._clamp(self.max, -1, self.max_open) # type: ignore
+
+ if lt_min or gt_max:
+ self.fail(
+ _("{value} is not in the range {range}.").format(
+ value=rv, range=self._describe_range()
+ ),
+ param,
+ ctx,
+ )
+
+ return rv
+
+ def _clamp(self, bound: float, dir: "te.Literal[1, -1]", open: bool) -> float:
"""Find the valid value to clamp to bound in the given
direction.
@@ -241,23 +484,33 @@ class _NumberRangeBase(_NumberParamTypeBase):
:param dir: 1 or -1 indicating the direction to move.
:param open: If true, the range does not include the bound.
"""
- pass
+ raise NotImplementedError
- def _describe_range(self) ->str:
+ def _describe_range(self) -> str:
"""Describe the range for use in help text."""
- pass
+ if self.min is None:
+ op = "<" if self.max_open else "<="
+ return f"x{op}{self.max}"
+
+ if self.max is None:
+ op = ">" if self.min_open else ">="
+ return f"x{op}{self.min}"
- def __repr__(self) ->str:
- clamp = ' clamped' if self.clamp else ''
- return f'<{type(self).__name__} {self._describe_range()}{clamp}>'
+ lop = "<" if self.min_open else "<="
+ rop = "<" if self.max_open else "<="
+ return f"{self.min}{lop}x{rop}{self.max}"
+
+ def __repr__(self) -> str:
+ clamp = " clamped" if self.clamp else ""
+ return f"<{type(self).__name__} {self._describe_range()}{clamp}>"
class IntParamType(_NumberParamTypeBase):
- name = 'integer'
+ name = "integer"
_number_class = int
- def __repr__(self) ->str:
- return 'INT'
+ def __repr__(self) -> str:
+ return "INT"
class IntRange(_NumberRangeBase, IntParamType):
@@ -274,15 +527,24 @@ class IntRange(_NumberRangeBase, IntParamType):
.. versionchanged:: 8.0
Added the ``min_open`` and ``max_open`` parameters.
"""
- name = 'integer range'
+
+ name = "integer range"
+
+ def _clamp( # type: ignore
+ self, bound: int, dir: "te.Literal[1, -1]", open: bool
+ ) -> int:
+ if not open:
+ return bound
+
+ return bound + dir
class FloatParamType(_NumberParamTypeBase):
- name = 'float'
+ name = "float"
_number_class = float
- def __repr__(self) ->str:
- return 'FLOAT'
+ def __repr__(self) -> str:
+ return "FLOAT"
class FloatRange(_NumberRangeBase, FloatParamType):
@@ -300,29 +562,81 @@ class FloatRange(_NumberRangeBase, FloatParamType):
.. versionchanged:: 8.0
Added the ``min_open`` and ``max_open`` parameters.
"""
- name = 'float range'
- def __init__(self, min: t.Optional[float]=None, max: t.Optional[float]=
- None, min_open: bool=False, max_open: bool=False, clamp: bool=False
- ) ->None:
- super().__init__(min=min, max=max, min_open=min_open, max_open=
- max_open, clamp=clamp)
+ name = "float range"
+
+ def __init__(
+ self,
+ min: t.Optional[float] = None,
+ max: t.Optional[float] = None,
+ min_open: bool = False,
+ max_open: bool = False,
+ clamp: bool = False,
+ ) -> None:
+ super().__init__(
+ min=min, max=max, min_open=min_open, max_open=max_open, clamp=clamp
+ )
+
if (min_open or max_open) and clamp:
- raise TypeError('Clamping is not supported for open bounds.')
+ raise TypeError("Clamping is not supported for open bounds.")
+
+ def _clamp(self, bound: float, dir: "te.Literal[1, -1]", open: bool) -> float:
+ if not open:
+ return bound
+
+ # Could use Python 3.9's math.nextafter here, but clamping an
+ # open float range doesn't seem to be particularly useful. It's
+ # left up to the user to write a callback to do it if needed.
+ raise RuntimeError("Clamping is not supported for open bounds.")
class BoolParamType(ParamType):
- name = 'boolean'
+ name = "boolean"
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ if value in {False, True}:
+ return bool(value)
+
+ norm = value.strip().lower()
+
+ if norm in {"1", "true", "t", "yes", "y", "on"}:
+ return True
+
+ if norm in {"0", "false", "f", "no", "n", "off"}:
+ return False
- def __repr__(self) ->str:
- return 'BOOL'
+ self.fail(
+ _("{value!r} is not a valid boolean.").format(value=value), param, ctx
+ )
+
+ def __repr__(self) -> str:
+ return "BOOL"
class UUIDParameterType(ParamType):
- name = 'uuid'
+ name = "uuid"
+
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ import uuid
+
+ if isinstance(value, uuid.UUID):
+ return value
+
+ value = value.strip()
+
+ try:
+ return uuid.UUID(value)
+ except ValueError:
+ self.fail(
+ _("{value!r} is not a valid UUID.").format(value=value), param, ctx
+ )
- def __repr__(self) ->str:
- return 'UUID'
+ def __repr__(self) -> str:
+ return "UUID"
class File(ParamType):
@@ -351,20 +665,84 @@ class File(ParamType):
See :ref:`file-args` for more information.
"""
- name = 'filename'
+
+ name = "filename"
envvar_list_splitter: t.ClassVar[str] = os.path.pathsep
- def __init__(self, mode: str='r', encoding: t.Optional[str]=None,
- errors: t.Optional[str]='strict', lazy: t.Optional[bool]=None,
- atomic: bool=False) ->None:
+ def __init__(
+ self,
+ mode: str = "r",
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+ lazy: t.Optional[bool] = None,
+ atomic: bool = False,
+ ) -> None:
self.mode = mode
self.encoding = encoding
self.errors = errors
self.lazy = lazy
self.atomic = atomic
- def shell_complete(self, ctx: 'Context', param: 'Parameter', incomplete:
- str) ->t.List['CompletionItem']:
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict.update(mode=self.mode, encoding=self.encoding)
+ return info_dict
+
+ def resolve_lazy_flag(self, value: "t.Union[str, os.PathLike[str]]") -> bool:
+ if self.lazy is not None:
+ return self.lazy
+ if os.fspath(value) == "-":
+ return False
+ elif "w" in self.mode:
+ return True
+ return False
+
+ def convert(
+ self,
+ value: t.Union[str, "os.PathLike[str]", t.IO[t.Any]],
+ param: t.Optional["Parameter"],
+ ctx: t.Optional["Context"],
+ ) -> t.IO[t.Any]:
+ if _is_file_like(value):
+ return value
+
+ value = t.cast("t.Union[str, os.PathLike[str]]", value)
+
+ try:
+ lazy = self.resolve_lazy_flag(value)
+
+ if lazy:
+ lf = LazyFile(
+ value, self.mode, self.encoding, self.errors, atomic=self.atomic
+ )
+
+ if ctx is not None:
+ ctx.call_on_close(lf.close_intelligently)
+
+ return t.cast(t.IO[t.Any], lf)
+
+ f, should_close = open_stream(
+ value, self.mode, self.encoding, self.errors, atomic=self.atomic
+ )
+
+ # If a context is provided, we automatically close the file
+ # at the end of the context execution (or flush out). If a
+ # context does not exist, it's the caller's responsibility to
+ # properly close the file. This for instance happens when the
+ # type is used with prompts.
+ if ctx is not None:
+ if should_close:
+ ctx.call_on_close(safecall(f.close))
+ else:
+ ctx.call_on_close(safecall(f.flush))
+
+ return f
+ except OSError as e: # noqa: B014
+ self.fail(f"'{format_filename(value)}': {e.strerror}", param, ctx)
+
+ def shell_complete(
+ self, ctx: "Context", param: "Parameter", incomplete: str
+ ) -> t.List["CompletionItem"]:
"""Return a special completion marker that tells the completion
system to use the shell to provide file path completions.
@@ -374,7 +752,13 @@ class File(ParamType):
.. versionadded:: 8.0
"""
- pass
+ from click.shell_completion import CompletionItem
+
+ return [CompletionItem(incomplete, type="file")]
+
+
+def _is_file_like(value: t.Any) -> "te.TypeGuard[t.IO[t.Any]]":
+ return hasattr(value, "read") or hasattr(value, "write")
class Path(ParamType):
@@ -409,12 +793,21 @@ class Path(ParamType):
.. versionchanged:: 6.0
Added the ``allow_dash`` parameter.
"""
+
envvar_list_splitter: t.ClassVar[str] = os.path.pathsep
- def __init__(self, exists: bool=False, file_okay: bool=True, dir_okay:
- bool=True, writable: bool=False, readable: bool=True, resolve_path:
- bool=False, allow_dash: bool=False, path_type: t.Optional[t.Type[t.
- Any]]=None, executable: bool=False):
+ def __init__(
+ self,
+ exists: bool = False,
+ file_okay: bool = True,
+ dir_okay: bool = True,
+ writable: bool = False,
+ readable: bool = True,
+ resolve_path: bool = False,
+ allow_dash: bool = False,
+ path_type: t.Optional[t.Type[t.Any]] = None,
+ executable: bool = False,
+ ):
self.exists = exists
self.file_okay = file_okay
self.dir_okay = dir_okay
@@ -424,15 +817,119 @@ class Path(ParamType):
self.resolve_path = resolve_path
self.allow_dash = allow_dash
self.type = path_type
+
if self.file_okay and not self.dir_okay:
- self.name: str = _('file')
+ self.name: str = _("file")
elif self.dir_okay and not self.file_okay:
- self.name = _('directory')
+ self.name = _("directory")
else:
- self.name = _('path')
-
- def shell_complete(self, ctx: 'Context', param: 'Parameter', incomplete:
- str) ->t.List['CompletionItem']:
+ self.name = _("path")
+
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict.update(
+ exists=self.exists,
+ file_okay=self.file_okay,
+ dir_okay=self.dir_okay,
+ writable=self.writable,
+ readable=self.readable,
+ allow_dash=self.allow_dash,
+ )
+ return info_dict
+
+ def coerce_path_result(
+ self, value: "t.Union[str, os.PathLike[str]]"
+ ) -> "t.Union[str, bytes, os.PathLike[str]]":
+ if self.type is not None and not isinstance(value, self.type):
+ if self.type is str:
+ return os.fsdecode(value)
+ elif self.type is bytes:
+ return os.fsencode(value)
+ else:
+ return t.cast("os.PathLike[str]", self.type(value))
+
+ return value
+
+ def convert(
+ self,
+ value: "t.Union[str, os.PathLike[str]]",
+ param: t.Optional["Parameter"],
+ ctx: t.Optional["Context"],
+ ) -> "t.Union[str, bytes, os.PathLike[str]]":
+ rv = value
+
+ is_dash = self.file_okay and self.allow_dash and rv in (b"-", "-")
+
+ if not is_dash:
+ if self.resolve_path:
+ # os.path.realpath doesn't resolve symlinks on Windows
+ # until Python 3.8. Use pathlib for now.
+ import pathlib
+
+ rv = os.fsdecode(pathlib.Path(rv).resolve())
+
+ try:
+ st = os.stat(rv)
+ except OSError:
+ if not self.exists:
+ return self.coerce_path_result(rv)
+ self.fail(
+ _("{name} {filename!r} does not exist.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+
+ if not self.file_okay and stat.S_ISREG(st.st_mode):
+ self.fail(
+ _("{name} {filename!r} is a file.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+ if not self.dir_okay and stat.S_ISDIR(st.st_mode):
+ self.fail(
+ _("{name} '{filename}' is a directory.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+
+ if self.readable and not os.access(rv, os.R_OK):
+ self.fail(
+ _("{name} {filename!r} is not readable.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+
+ if self.writable and not os.access(rv, os.W_OK):
+ self.fail(
+ _("{name} {filename!r} is not writable.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+
+ if self.executable and not os.access(value, os.X_OK):
+ self.fail(
+ _("{name} {filename!r} is not executable.").format(
+ name=self.name.title(), filename=format_filename(value)
+ ),
+ param,
+ ctx,
+ )
+
+ return self.coerce_path_result(rv)
+
+ def shell_complete(
+ self, ctx: "Context", param: "Parameter", incomplete: str
+ ) -> t.List["CompletionItem"]:
"""Return a special completion marker that tells the completion
system to use the shell to provide path completions for only
directories or any paths.
@@ -443,7 +940,10 @@ class Path(ParamType):
.. versionadded:: 8.0
"""
- pass
+ from click.shell_completion import CompletionItem
+
+ type = "dir" if self.dir_okay and not self.file_okay else "file"
+ return [CompletionItem(incomplete, type=type)]
class Tuple(CompositeParamType):
@@ -460,23 +960,130 @@ class Tuple(CompositeParamType):
:param types: a list of types that should be used for the tuple items.
"""
- def __init__(self, types: t.Sequence[t.Union[t.Type[t.Any], ParamType]]
- ) ->None:
+ def __init__(self, types: t.Sequence[t.Union[t.Type[t.Any], ParamType]]) -> None:
self.types: t.Sequence[ParamType] = [convert_type(ty) for ty in types]
+ def to_info_dict(self) -> t.Dict[str, t.Any]:
+ info_dict = super().to_info_dict()
+ info_dict["types"] = [t.to_info_dict() for t in self.types]
+ return info_dict
+
+ @property
+ def name(self) -> str: # type: ignore
+ return f"<{' '.join(ty.name for ty in self.types)}>"
+
+ @property
+ def arity(self) -> int: # type: ignore
+ return len(self.types)
-def convert_type(ty: t.Optional[t.Any], default: t.Optional[t.Any]=None
- ) ->ParamType:
+ def convert(
+ self, value: t.Any, param: t.Optional["Parameter"], ctx: t.Optional["Context"]
+ ) -> t.Any:
+ len_type = len(self.types)
+ len_value = len(value)
+
+ if len_value != len_type:
+ self.fail(
+ ngettext(
+ "{len_type} values are required, but {len_value} was given.",
+ "{len_type} values are required, but {len_value} were given.",
+ len_value,
+ ).format(len_type=len_type, len_value=len_value),
+ param=param,
+ ctx=ctx,
+ )
+
+ return tuple(ty(x, param, ctx) for ty, x in zip(self.types, value))
+
+
+def convert_type(ty: t.Optional[t.Any], default: t.Optional[t.Any] = None) -> ParamType:
"""Find the most appropriate :class:`ParamType` for the given Python
type. If the type isn't provided, it can be inferred from a default
value.
"""
- pass
+ guessed_type = False
+
+ if ty is None and default is not None:
+ if isinstance(default, (tuple, list)):
+ # If the default is empty, ty will remain None and will
+ # return STRING.
+ if default:
+ item = default[0]
+
+ # A tuple of tuples needs to detect the inner types.
+ # Can't call convert recursively because that would
+ # incorrectly unwind the tuple to a single type.
+ if isinstance(item, (tuple, list)):
+ ty = tuple(map(type, item))
+ else:
+ ty = type(item)
+ else:
+ ty = type(default)
+
+ guessed_type = True
+
+ if isinstance(ty, tuple):
+ return Tuple(ty)
+
+ if isinstance(ty, ParamType):
+ return ty
+
+ if ty is str or ty is None:
+ return STRING
+
+ if ty is int:
+ return INT
+ if ty is float:
+ return FLOAT
+ if ty is bool:
+ return BOOL
+
+ if guessed_type:
+ return STRING
+
+ if __debug__:
+ try:
+ if issubclass(ty, ParamType):
+ raise AssertionError(
+ f"Attempted to use an uninstantiated parameter type ({ty})."
+ )
+ except TypeError:
+ # ty is an instance (correct), so issubclass fails.
+ pass
+
+ return FuncParamType(ty)
+
+
+#: A dummy parameter type that just does nothing. From a user's
+#: perspective this appears to just be the same as `STRING` but
+#: internally no string conversion takes place if the input was bytes.
+#: This is usually useful when working with file paths as they can
+#: appear in bytes and unicode.
+#:
+#: For path related uses the :class:`Path` type is a better choice but
+#: there are situations where an unprocessed type is useful which is why
+#: it is is provided.
+#:
+#: .. versionadded:: 4.0
UNPROCESSED = UnprocessedParamType()
+
+#: A unicode string parameter type which is the implicit default. This
+#: can also be selected by using ``str`` as type.
STRING = StringParamType()
+
+#: An integer parameter. This can also be selected by using ``int`` as
+#: type.
INT = IntParamType()
+
+#: A floating point value parameter. This can also be selected by using
+#: ``float`` as type.
FLOAT = FloatParamType()
+
+#: A boolean parameter. This is the default for boolean flags. This can
+#: also be selected by using ``bool`` as a type.
BOOL = BoolParamType()
+
+#: A UUID parameter.
UUID = UUIDParameterType()
diff --git a/src/click/utils.py b/src/click/utils.py
index 0b2575c..d536434 100644
--- a/src/click/utils.py
+++ b/src/click/utils.py
@@ -5,6 +5,7 @@ import typing as t
from functools import update_wrapper
from types import ModuleType
from types import TracebackType
+
from ._compat import _default_text_stderr
from ._compat import _default_text_stdout
from ._compat import _find_binary_writer
@@ -16,25 +17,90 @@ from ._compat import strip_ansi
from ._compat import text_streams
from ._compat import WIN
from .globals import resolve_color_default
+
if t.TYPE_CHECKING:
import typing_extensions as te
- P = te.ParamSpec('P')
-R = t.TypeVar('R')
+
+ P = te.ParamSpec("P")
+
+R = t.TypeVar("R")
-def safecall(func: 't.Callable[P, R]') ->'t.Callable[P, t.Optional[R]]':
+def _posixify(name: str) -> str:
+ return "-".join(name.split()).lower()
+
+
+def safecall(func: "t.Callable[P, R]") -> "t.Callable[P, t.Optional[R]]":
"""Wraps a function so that it swallows exceptions."""
- pass
+ def wrapper(*args: "P.args", **kwargs: "P.kwargs") -> t.Optional[R]:
+ try:
+ return func(*args, **kwargs)
+ except Exception:
+ pass
+ return None
+
+ return update_wrapper(wrapper, func)
-def make_str(value: t.Any) ->str:
+
+def make_str(value: t.Any) -> str:
"""Converts a value into a valid string."""
- pass
+ if isinstance(value, bytes):
+ try:
+ return value.decode(sys.getfilesystemencoding())
+ except UnicodeError:
+ return value.decode("utf-8", "replace")
+ return str(value)
-def make_default_short_help(help: str, max_length: int=45) ->str:
+def make_default_short_help(help: str, max_length: int = 45) -> str:
"""Returns a condensed version of help string."""
- pass
+ # Consider only the first paragraph.
+ paragraph_end = help.find("\n\n")
+
+ if paragraph_end != -1:
+ help = help[:paragraph_end]
+
+ # Collapse newlines, tabs, and spaces.
+ words = help.split()
+
+ if not words:
+ return ""
+
+ # The first paragraph started with a "no rewrap" marker, ignore it.
+ if words[0] == "\b":
+ words = words[1:]
+
+ total_length = 0
+ last_index = len(words) - 1
+
+ for i, word in enumerate(words):
+ total_length += len(word) + (i > 0)
+
+ if total_length > max_length: # too long, truncate
+ break
+
+ if word[-1] == ".": # sentence end, truncate without "..."
+ return " ".join(words[: i + 1])
+
+ if total_length == max_length and i != last_index:
+ break # not at sentence end, truncate with "..."
+ else:
+ return " ".join(words) # no truncation needed
+
+ # Account for the length of the suffix.
+ total_length += len("...")
+
+ # remove words until the length is short enough
+ while i > 0:
+ total_length -= len(words[i]) + (i > 0)
+
+ if total_length <= max_length:
+ break
+
+ i -= 1
+
+ return " ".join(words[:i]) + "..."
class LazyFile:
@@ -44,9 +110,14 @@ class LazyFile:
files for writing.
"""
- def __init__(self, filename: t.Union[str, 'os.PathLike[str]'], mode:
- str='r', encoding: t.Optional[str]=None, errors: t.Optional[str]=
- 'strict', atomic: bool=False):
+ def __init__(
+ self,
+ filename: t.Union[str, "os.PathLike[str]"],
+ mode: str = "r",
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+ atomic: bool = False,
+ ):
self.name: str = os.fspath(filename)
self.mode = mode
self.encoding = encoding
@@ -54,78 +125,104 @@ class LazyFile:
self.atomic = atomic
self._f: t.Optional[t.IO[t.Any]]
self.should_close: bool
- if self.name == '-':
- self._f, self.should_close = open_stream(filename, mode,
- encoding, errors)
+
+ if self.name == "-":
+ self._f, self.should_close = open_stream(filename, mode, encoding, errors)
else:
- if 'r' in mode:
+ if "r" in mode:
+ # Open and close the file in case we're opening it for
+ # reading so that we can catch at least some errors in
+ # some cases early.
open(filename, mode).close()
self._f = None
self.should_close = True
- def __getattr__(self, name: str) ->t.Any:
+ def __getattr__(self, name: str) -> t.Any:
return getattr(self.open(), name)
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
if self._f is not None:
return repr(self._f)
return f"<unopened file '{format_filename(self.name)}' {self.mode}>"
- def open(self) ->t.IO[t.Any]:
+ def open(self) -> t.IO[t.Any]:
"""Opens the file if it's not yet open. This call might fail with
a :exc:`FileError`. Not handling this error will produce an error
that Click shows.
"""
- pass
-
- def close(self) ->None:
+ if self._f is not None:
+ return self._f
+ try:
+ rv, self.should_close = open_stream(
+ self.name, self.mode, self.encoding, self.errors, atomic=self.atomic
+ )
+ except OSError as e: # noqa: E402
+ from .exceptions import FileError
+
+ raise FileError(self.name, hint=e.strerror) from e
+ self._f = rv
+ return rv
+
+ def close(self) -> None:
"""Closes the underlying file, no matter what."""
- pass
+ if self._f is not None:
+ self._f.close()
- def close_intelligently(self) ->None:
+ def close_intelligently(self) -> None:
"""This function only closes the file if it was opened by the lazy
file wrapper. For instance this will never close stdin.
"""
- pass
+ if self.should_close:
+ self.close()
- def __enter__(self) ->'LazyFile':
+ def __enter__(self) -> "LazyFile":
return self
- def __exit__(self, exc_type: t.Optional[t.Type[BaseException]],
- exc_value: t.Optional[BaseException], tb: t.Optional[TracebackType]
- ) ->None:
+ def __exit__(
+ self,
+ exc_type: t.Optional[t.Type[BaseException]],
+ exc_value: t.Optional[BaseException],
+ tb: t.Optional[TracebackType],
+ ) -> None:
self.close_intelligently()
- def __iter__(self) ->t.Iterator[t.AnyStr]:
+ def __iter__(self) -> t.Iterator[t.AnyStr]:
self.open()
- return iter(self._f)
+ return iter(self._f) # type: ignore
class KeepOpenFile:
-
- def __init__(self, file: t.IO[t.Any]) ->None:
+ def __init__(self, file: t.IO[t.Any]) -> None:
self._file: t.IO[t.Any] = file
- def __getattr__(self, name: str) ->t.Any:
+ def __getattr__(self, name: str) -> t.Any:
return getattr(self._file, name)
- def __enter__(self) ->'KeepOpenFile':
+ def __enter__(self) -> "KeepOpenFile":
return self
- def __exit__(self, exc_type: t.Optional[t.Type[BaseException]],
- exc_value: t.Optional[BaseException], tb: t.Optional[TracebackType]
- ) ->None:
+ def __exit__(
+ self,
+ exc_type: t.Optional[t.Type[BaseException]],
+ exc_value: t.Optional[BaseException],
+ tb: t.Optional[TracebackType],
+ ) -> None:
pass
- def __repr__(self) ->str:
+ def __repr__(self) -> str:
return repr(self._file)
- def __iter__(self) ->t.Iterator[t.AnyStr]:
+ def __iter__(self) -> t.Iterator[t.AnyStr]:
return iter(self._file)
-def echo(message: t.Optional[t.Any]=None, file: t.Optional[t.IO[t.Any]]=
- None, nl: bool=True, err: bool=False, color: t.Optional[bool]=None) ->None:
+def echo(
+ message: t.Optional[t.Any] = None,
+ file: t.Optional[t.IO[t.Any]] = None,
+ nl: bool = True,
+ err: bool = False,
+ color: t.Optional[bool] = None,
+) -> None:
"""Print a message and newline to stdout or a file. This should be
used instead of :func:`print` because it provides better support
for different data, files, and environments.
@@ -164,22 +261,81 @@ def echo(message: t.Optional[t.Any]=None, file: t.Optional[t.IO[t.Any]]=
.. versionchanged:: 2.0
Support colors on Windows if colorama is installed.
"""
- pass
-
-
-def get_binary_stream(name: "te.Literal['stdin', 'stdout', 'stderr']"
- ) ->t.BinaryIO:
+ if file is None:
+ if err:
+ file = _default_text_stderr()
+ else:
+ file = _default_text_stdout()
+
+ # There are no standard streams attached to write to. For example,
+ # pythonw on Windows.
+ if file is None:
+ return
+
+ # Convert non bytes/text into the native string type.
+ if message is not None and not isinstance(message, (str, bytes, bytearray)):
+ out: t.Optional[t.Union[str, bytes]] = str(message)
+ else:
+ out = message
+
+ if nl:
+ out = out or ""
+ if isinstance(out, str):
+ out += "\n"
+ else:
+ out += b"\n"
+
+ if not out:
+ file.flush()
+ return
+
+ # If there is a message and the value looks like bytes, we manually
+ # need to find the binary stream and write the message in there.
+ # This is done separately so that most stream types will work as you
+ # would expect. Eg: you can write to StringIO for other cases.
+ if isinstance(out, (bytes, bytearray)):
+ binary_file = _find_binary_writer(file)
+
+ if binary_file is not None:
+ file.flush()
+ binary_file.write(out)
+ binary_file.flush()
+ return
+
+ # ANSI style code support. For no message or bytes, nothing happens.
+ # When outputting to a file instead of a terminal, strip codes.
+ else:
+ color = resolve_color_default(color)
+
+ if should_strip_ansi(file, color):
+ out = strip_ansi(out)
+ elif WIN:
+ if auto_wrap_for_ansi is not None:
+ file = auto_wrap_for_ansi(file) # type: ignore
+ elif not color:
+ out = strip_ansi(out)
+
+ file.write(out) # type: ignore
+ file.flush()
+
+
+def get_binary_stream(name: "te.Literal['stdin', 'stdout', 'stderr']") -> t.BinaryIO:
"""Returns a system stream for byte processing.
:param name: the name of the stream to open. Valid names are ``'stdin'``,
``'stdout'`` and ``'stderr'``
"""
- pass
+ opener = binary_streams.get(name)
+ if opener is None:
+ raise TypeError(f"Unknown standard stream '{name}'")
+ return opener()
-def get_text_stream(name: "te.Literal['stdin', 'stdout', 'stderr']",
- encoding: t.Optional[str]=None, errors: t.Optional[str]='strict'
- ) ->t.TextIO:
+def get_text_stream(
+ name: "te.Literal['stdin', 'stdout', 'stderr']",
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+) -> t.TextIO:
"""Returns a system stream for text processing. This usually returns
a wrapped stream around a binary stream returned from
:func:`get_binary_stream` but it also can take shortcuts for already
@@ -190,12 +346,20 @@ def get_text_stream(name: "te.Literal['stdin', 'stdout', 'stderr']",
:param encoding: overrides the detected default encoding.
:param errors: overrides the default error mode.
"""
- pass
-
-
-def open_file(filename: str, mode: str='r', encoding: t.Optional[str]=None,
- errors: t.Optional[str]='strict', lazy: bool=False, atomic: bool=False
- ) ->t.IO[t.Any]:
+ opener = text_streams.get(name)
+ if opener is None:
+ raise TypeError(f"Unknown standard stream '{name}'")
+ return opener(encoding, errors)
+
+
+def open_file(
+ filename: str,
+ mode: str = "r",
+ encoding: t.Optional[str] = None,
+ errors: t.Optional[str] = "strict",
+ lazy: bool = False,
+ atomic: bool = False,
+) -> t.IO[t.Any]:
"""Open a file, with extra behavior to handle ``'-'`` to indicate
a standard stream, lazy open on write, and atomic write. Similar to
the behavior of the :class:`~click.File` param type.
@@ -224,12 +388,23 @@ def open_file(filename: str, mode: str='r', encoding: t.Optional[str]=None,
.. versionadded:: 3.0
"""
- pass
+ if lazy:
+ return t.cast(
+ t.IO[t.Any], LazyFile(filename, mode, encoding, errors, atomic=atomic)
+ )
+
+ f, should_close = open_stream(filename, mode, encoding, errors, atomic=atomic)
+
+ if not should_close:
+ f = t.cast(t.IO[t.Any], KeepOpenFile(f))
+
+ return f
-def format_filename(filename:
- 't.Union[str, bytes, os.PathLike[str], os.PathLike[bytes]]', shorten:
- bool=False) ->str:
+def format_filename(
+ filename: "t.Union[str, bytes, os.PathLike[str], os.PathLike[bytes]]",
+ shorten: bool = False,
+) -> str:
"""Format a filename as a string for display. Ensures the filename can be
displayed by replacing any invalid bytes or surrogate escapes in the name
with the replacement character ``�``.
@@ -253,12 +428,23 @@ def format_filename(filename:
:param shorten: this optionally shortens the filename to strip of the
path that leads up to it.
"""
- pass
+ if shorten:
+ filename = os.path.basename(filename)
+ else:
+ filename = os.fspath(filename)
+ if isinstance(filename, bytes):
+ filename = filename.decode(sys.getfilesystemencoding(), "replace")
+ else:
+ filename = filename.encode("utf-8", "surrogateescape").decode(
+ "utf-8", "replace"
+ )
-def get_app_dir(app_name: str, roaming: bool=True, force_posix: bool=False
- ) ->str:
- """Returns the config folder for the application. The default behavior
+ return filename
+
+
+def get_app_dir(app_name: str, roaming: bool = True, force_posix: bool = False) -> str:
+ r"""Returns the config folder for the application. The default behavior
is to return whatever is most appropriate for the operating system.
To give you an idea, for an app called ``"Foo Bar"``, something like
@@ -273,9 +459,9 @@ def get_app_dir(app_name: str, roaming: bool=True, force_posix: bool=False
Unix (POSIX):
``~/.foo-bar``
Windows (roaming):
- ``C:\\Users\\<user>\\AppData\\Roaming\\Foo Bar``
+ ``C:\Users\<user>\AppData\Roaming\Foo Bar``
Windows (not roaming):
- ``C:\\Users\\<user>\\AppData\\Local\\Foo Bar``
+ ``C:\Users\<user>\AppData\Local\Foo Bar``
.. versionadded:: 2.0
@@ -288,7 +474,22 @@ def get_app_dir(app_name: str, roaming: bool=True, force_posix: bool=False
dot instead of the XDG config home or darwin's
application support folder.
"""
- pass
+ if WIN:
+ key = "APPDATA" if roaming else "LOCALAPPDATA"
+ folder = os.environ.get(key)
+ if folder is None:
+ folder = os.path.expanduser("~")
+ return os.path.join(folder, app_name)
+ if force_posix:
+ return os.path.join(os.path.expanduser(f"~/.{_posixify(app_name)}"))
+ if sys.platform == "darwin":
+ return os.path.join(
+ os.path.expanduser("~/Library/Application Support"), app_name
+ )
+ return os.path.join(
+ os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")),
+ _posixify(app_name),
+ )
class PacifyFlushWrapper:
@@ -300,15 +501,25 @@ class PacifyFlushWrapper:
pipe, all calls and attributes are proxied.
"""
- def __init__(self, wrapped: t.IO[t.Any]) ->None:
+ def __init__(self, wrapped: t.IO[t.Any]) -> None:
self.wrapped = wrapped
- def __getattr__(self, attr: str) ->t.Any:
+ def flush(self) -> None:
+ try:
+ self.wrapped.flush()
+ except OSError as e:
+ import errno
+
+ if e.errno != errno.EPIPE:
+ raise
+
+ def __getattr__(self, attr: str) -> t.Any:
return getattr(self.wrapped, attr)
-def _detect_program_name(path: t.Optional[str]=None, _main: t.Optional[
- ModuleType]=None) ->str:
+def _detect_program_name(
+ path: t.Optional[str] = None, _main: t.Optional[ModuleType] = None
+) -> str:
"""Determine the command used to run the program, for use in help
text. If a file or entry point was executed, the file name is
returned. If ``python -m`` was used to execute a module or package,
@@ -329,11 +540,45 @@ def _detect_program_name(path: t.Optional[str]=None, _main: t.Optional[
:meta private:
"""
- pass
-
-
-def _expand_args(args: t.Iterable[str], *, user: bool=True, env: bool=True,
- glob_recursive: bool=True) ->t.List[str]:
+ if _main is None:
+ _main = sys.modules["__main__"]
+
+ if not path:
+ path = sys.argv[0]
+
+ # The value of __package__ indicates how Python was called. It may
+ # not exist if a setuptools script is installed as an egg. It may be
+ # set incorrectly for entry points created with pip on Windows.
+ # It is set to "" inside a Shiv or PEX zipapp.
+ if getattr(_main, "__package__", None) in {None, ""} or (
+ os.name == "nt"
+ and _main.__package__ == ""
+ and not os.path.exists(path)
+ and os.path.exists(f"{path}.exe")
+ ):
+ # Executed a file, like "python app.py".
+ return os.path.basename(path)
+
+ # Executed a module, like "python -m example".
+ # Rewritten by Python from "-m script" to "/path/to/script.py".
+ # Need to look at main module to determine how it was executed.
+ py_module = t.cast(str, _main.__package__)
+ name = os.path.splitext(os.path.basename(path))[0]
+
+ # A submodule like "example.cli".
+ if name != "__main__":
+ py_module = f"{py_module}.{name}"
+
+ return f"python -m {py_module.lstrip('.')}"
+
+
+def _expand_args(
+ args: t.Iterable[str],
+ *,
+ user: bool = True,
+ env: bool = True,
+ glob_recursive: bool = True,
+) -> t.List[str]:
"""Simulate Unix shell expansion with Python functions.
See :func:`glob.glob`, :func:`os.path.expanduser`, and
@@ -355,4 +600,25 @@ def _expand_args(args: t.Iterable[str], *, user: bool=True, env: bool=True,
:meta private:
"""
- pass
+ from glob import glob
+
+ out = []
+
+ for arg in args:
+ if user:
+ arg = os.path.expanduser(arg)
+
+ if env:
+ arg = os.path.expandvars(arg)
+
+ try:
+ matches = glob(arg, recursive=glob_recursive)
+ except re.error:
+ matches = []
+
+ if not matches:
+ out.append(arg)
+ else:
+ out.extend(matches)
+
+ return out