feat(08): implement SSOT architecture for network configuration

This commit is contained in:
2026-04-23 13:58:33 +03:00
parent 5978917494
commit a5a460e765
1039 changed files with 138862 additions and 105259 deletions

View File

@@ -14,6 +14,7 @@ from .core import Context as Context
from .core import Group as Group
from .core import Option as Option
from .core import Parameter as Parameter
from .core import ParameterSource as ParameterSource
from .decorators import argument as argument
from .decorators import command as command
from .decorators import confirmation_option as confirmation_option

View File

@@ -378,8 +378,11 @@ def pager(generator: cabc.Iterable[str], color: bool | None = None) -> None:
if not isatty(sys.stdin) or not isatty(stdout):
return _nullpager(stdout, generator, color)
# Split and normalize the pager command into parts.
pager_cmd_parts = shlex.split(os.environ.get("PAGER", ""), posix=False)
# Split using POSIX mode (the default) so that quote characters are
# stripped from tokens and quoted Windows paths are preserved.
# Non-POSIX mode retains quotes in tokens, and wrapping tokens
# with shlex.quote re-introduces quoting issues on Windows.
pager_cmd_parts = shlex.split(os.environ.get("PAGER", ""))
if pager_cmd_parts:
if WIN:
if _tempfilepager(generator, pager_cmd_parts, color):
@@ -411,11 +414,19 @@ def pager(generator: cabc.Iterable[str], color: bool | None = None) -> None:
def _pipepager(
generator: cabc.Iterable[str], cmd_parts: list[str], color: bool | None
) -> bool:
"""Page through text by feeding it to another program. Invoking a
pager through this might support colors.
"""Page through text by feeding it to another program.
Returns `True` if the command was found, `False` otherwise and thus another
pager should be attempted.
Invokes the pager via :class:`subprocess.Popen` with an ``argv`` list
produced by :func:`shlex.split`. The command is resolved to an absolute
path with :func:`shutil.which` as recommended by the
:mod:`subprocess` docs for Windows compatibility.
Invoking a pager through this might support colors: if piping to
``less`` and the user hasn't decided on colors, ``LESS=-R`` is set
automatically.
Returns ``True`` if the command was found and executed, ``False``
otherwise so another pager can be attempted.
"""
# Split the command into the invoked CLI and its parameters.
if not cmd_parts:
@@ -509,8 +520,13 @@ def _tempfilepager(
) -> bool:
"""Page through text by invoking a program on a temporary file.
Returns `True` if the command was found, `False` otherwise and thus another
pager should be attempted.
Used as the primary pager strategy on Windows (where piping to
``more`` adds spurious ``\\r\\n``), and as a fallback on other
platforms. The command is resolved to an absolute path with
:func:`shutil.which`.
Returns ``True`` if the command was found and executed, ``False``
otherwise so another pager can be attempted.
"""
# Split the command into the invoked CLI and its parameters.
if not cmd_parts:
@@ -592,6 +608,8 @@ class Editor:
return "vi"
def edit_files(self, filenames: cabc.Iterable[str]) -> None:
"""Open files in the user's editor."""
import shlex
import subprocess
editor = self.get_editor()
@@ -601,11 +619,13 @@ class Editor:
environ = os.environ.copy()
environ.update(self.env)
exc_filename = " ".join(f'"{filename}"' for filename in filenames)
try:
# Split in POSIX mode (the default) for the same reasons as
# in pager(): strips quotes from tokens and preserves quoted
# Windows paths.
c = subprocess.Popen(
args=f"{editor} {exc_filename}", env=environ, shell=True
args=shlex.split(editor) + list(filenames),
env=environ,
)
exit_code = c.wait()
if exit_code != 0:
@@ -754,8 +774,6 @@ def _translate_ch_to_exc(ch: str) -> None:
if ch == "\x1a" and WIN: # Windows, Ctrl+Z
raise EOFError()
return None
if sys.platform == "win32":
import msvcrt

View File

@@ -140,13 +140,26 @@ def iter_params_for_processing(
return sorted(declaration_order, key=sort_key)
class ParameterSource(enum.Enum):
"""This is an :class:`~enum.Enum` that indicates the source of a
class ParameterSource(enum.IntEnum):
"""This is an :class:`~enum.IntEnum` that indicates the source of a
parameter's value.
Use :meth:`click.Context.get_parameter_source` to get the
source for a parameter by name.
Members are ordered from most explicit to least explicit source.
This allows comparison to check if a value was explicitly provided:
.. code-block:: python
source = ctx.get_parameter_source("port")
if source < click.ParameterSource.DEFAULT_MAP:
... # value was explicitly set
.. versionchanged:: 8.3.3
Use :class:`~enum.IntEnum` and reorder members from most to
least explicit. Supports comparison operators.
.. versionchanged:: 8.0
Use :class:`~enum.Enum` and drop the ``validate`` method.
@@ -154,16 +167,16 @@ class ParameterSource(enum.Enum):
Added the ``PROMPT`` value.
"""
PROMPT = enum.auto()
"""Used a prompt to confirm a default or provide a value."""
COMMANDLINE = enum.auto()
"""The value was provided by the command line args."""
ENVIRONMENT = enum.auto()
"""The value was provided with an environment variable."""
DEFAULT = enum.auto()
"""Used the default specified by the parameter."""
DEFAULT_MAP = enum.auto()
"""Used a default provided by :attr:`Context.default_map`."""
PROMPT = enum.auto()
"""Used a prompt to confirm a default or provide a value."""
DEFAULT = enum.auto()
"""Used the default specified by the parameter."""
class Context:
@@ -685,6 +698,20 @@ class Context:
self.obj = rv = object_type()
return rv
def _default_map_has(self, name: str | None) -> bool:
"""Check if :attr:`default_map` contains a real value for ``name``.
Returns ``False`` when the key is absent, the map is ``None``,
``name`` is ``None``, or the stored value is the internal
:data:`UNSET` sentinel.
"""
return (
name is not None
and self.default_map is not None
and name in self.default_map
and self.default_map[name] is not UNSET
)
@t.overload
def lookup_default(
self, name: str, call: t.Literal[True] = True
@@ -705,15 +732,17 @@ class Context:
.. versionchanged:: 8.0
Added the ``call`` parameter.
"""
if self.default_map is not None:
value = self.default_map.get(name)
if not self._default_map_has(name):
return None
if call and callable(value):
return value()
# Assert to make the type checker happy.
assert self.default_map is not None
value = self.default_map[name]
return value
if call and callable(value):
return value()
return None
return value
def fail(self, message: str) -> t.NoReturn:
"""Aborts the execution of the program with a specific error
@@ -2281,9 +2310,7 @@ class Parameter:
name = self.name
value = ctx.lookup_default(name, call=False) if name is not None else None
if value is None and not (
ctx.default_map is not None and name is not None and name in ctx.default_map
):
if value is None and not ctx._default_map_has(name):
value = self.default
if call and callable(value):
@@ -2325,9 +2352,7 @@ class Parameter:
if value is UNSET:
default_map_value = ctx.lookup_default(self.name) # type: ignore[arg-type]
if default_map_value is not None or (
ctx.default_map is not None and self.name in ctx.default_map
):
if default_map_value is not None or ctx._default_map_has(self.name):
value = default_map_value
source = ParameterSource.DEFAULT_MAP
@@ -2563,7 +2588,7 @@ class Parameter:
if (
self.deprecated
and value is not UNSET
and source not in (ParameterSource.DEFAULT, ParameterSource.DEFAULT_MAP)
and source < ParameterSource.DEFAULT_MAP
):
extra_message = (
f" {self.deprecated}" if isinstance(self.deprecated, str) else ""
@@ -2891,13 +2916,30 @@ class Option(Parameter):
def get_default(
self, ctx: Context, call: bool = True
) -> t.Any | t.Callable[[], t.Any] | None:
"""Return the default value for this option.
For non-boolean flag options, ``default=True`` is treated as a sentinel
meaning "activate this flag by default" and is resolved to
:attr:`flag_value`. For example, with ``--upper/--lower`` feature
switches where ``flag_value="upper"`` and ``default=True``, the default
resolves to ``"upper"``.
.. caution::
This substitution only applies to non-boolean flags
(:attr:`is_bool_flag` is ``False``). For boolean flags, ``True`` is
a legitimate Python value and ``default=True`` is returned as-is.
.. versionchanged:: 8.3.3
``default=True`` is no longer substituted with ``flag_value`` for
boolean flags, fixing negative boolean flags like
``flag_value=False, default=True``.
"""
value = super().get_default(ctx, call=False)
# Lazily resolve default=True to flag_value. Doing this here
# (instead of eagerly in __init__) prevents callable flag_values
# (like classes) from being instantiated by the callable check below.
# https://github.com/pallets/click/issues/3121
if value is True and self.is_flag:
# Resolve default=True to flag_value lazily (here instead of
# __init__) to prevent callable flag_values (like classes) from
# being instantiated by the callable check below.
if value is True and self.is_flag and not self.is_bool_flag:
value = self.flag_value
elif call and callable(value):
value = value()
@@ -3110,7 +3152,7 @@ class Option(Parameter):
)[1]
elif self.is_bool_flag and not self.secondary_opts and not default_value:
default_string = ""
elif default_value == "":
elif isinstance(default_value, str) and default_value == "":
default_string = '""'
else:
default_string = str(default_value)
@@ -3164,10 +3206,10 @@ class Option(Parameter):
default = bool(default)
return confirm(self.prompt, default)
# If show_default is set to True/False, provide this to `prompt` as well. For
# non-bool values of `show_default`, we use `prompt`'s default behavior
# If show_default is given, provide this to `prompt` as well,
# otherwise we use `prompt`'s default behavior
prompt_kwargs: t.Any = {}
if isinstance(self.show_default, bool):
if self.show_default is not None:
prompt_kwargs["show_default"] = self.show_default
return prompt(
@@ -3285,7 +3327,7 @@ class Option(Parameter):
self.is_flag
and value is True
and not self.is_bool_flag
and source not in (ParameterSource.DEFAULT, ParameterSource.DEFAULT_MAP)
and source < ParameterSource.DEFAULT_MAP
):
value = self.flag_value
@@ -3295,7 +3337,7 @@ class Option(Parameter):
elif (
self.multiple
and value is not UNSET
and source not in (ParameterSource.DEFAULT, ParameterSource.DEFAULT_MAP)
and source < ParameterSource.DEFAULT_MAP
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]
@@ -3304,10 +3346,7 @@ class Option(Parameter):
# The value wasn't set, or used the param's default, prompt for one to the user
# if prompting is enabled.
elif (
(
value is UNSET
or source in (ParameterSource.DEFAULT, ParameterSource.DEFAULT_MAP)
)
(value is UNSET or source >= ParameterSource.DEFAULT_MAP)
and self.prompt is not None
and (self.required or self.prompt_required)
and not ctx.resilient_parsing

View File

@@ -60,7 +60,7 @@ def hidden_prompt_func(prompt: str) -> str:
def _build_prompt(
text: str,
suffix: str,
show_default: bool = False,
show_default: bool | str = False,
default: t.Any | None = None,
show_choices: bool = True,
type: ParamType | None = None,
@@ -68,6 +68,8 @@ def _build_prompt(
prompt = text
if type is not None and show_choices and isinstance(type, Choice):
prompt += f" ({', '.join(map(str, type.choices))})"
if isinstance(show_default, str):
default = f"({show_default})"
if default is not None and show_default:
prompt = f"{prompt} [{_format_default(default)}]"
return f"{prompt}{suffix}"
@@ -88,7 +90,7 @@ def prompt(
type: ParamType | t.Any | None = None,
value_proc: t.Callable[[str], t.Any] | None = None,
prompt_suffix: str = ": ",
show_default: bool = True,
show_default: bool | str = True,
err: bool = False,
show_choices: bool = True,
) -> t.Any:
@@ -112,6 +114,8 @@ def prompt(
convert a value.
:param prompt_suffix: a suffix that should be added to the prompt.
:param show_default: shows or hides the default value in the prompt.
If this value is a string, it shows that string
in parentheses instead of the actual value.
:param err: if set to true the file defaults to ``stderr`` instead of
``stdout``, the same as with echo.
:param show_choices: Show or hide choices if the passed type is a Choice.
@@ -119,6 +123,10 @@ def prompt(
show_choices is true and text is "Group by" then the
prompt will be "Group by (day, week): ".
.. versionchanged:: 8.3.3
``show_default`` can be a string to show a custom value instead
of the actual default, matching the help text behavior.
.. versionchanged:: 8.3.1
A space is no longer appended to the prompt.

View File

@@ -4,6 +4,7 @@ import collections.abc as cabc
import contextlib
import io
import os
import pdb
import shlex
import sys
import tempfile
@@ -100,21 +101,55 @@ class StreamMixer:
class _NamedTextIOWrapper(io.TextIOWrapper):
"""A :class:`~io.TextIOWrapper` with custom ``name`` and ``mode``
that does not close its underlying buffer.
An optional ``original_fd`` preserves the file descriptor of the
stream being replaced, so that C-level consumers that call
:meth:`fileno` (``faulthandler``, ``subprocess``, ...) still work.
Inspired by pytest's ``capsys``/``capfd`` split: see :doc:`/testing`
for details.
.. versionchanged:: 8.3.3
Added ``original_fd`` parameter and :meth:`fileno` override.
"""
def __init__(
self, buffer: t.BinaryIO, name: str, mode: str, **kwargs: t.Any
self,
buffer: t.BinaryIO,
name: str,
mode: str,
*,
original_fd: int = -1,
**kwargs: t.Any,
) -> None:
super().__init__(buffer, **kwargs)
self._name = name
self._mode = mode
self._original_fd = original_fd
def close(self) -> None:
"""
The buffer this object contains belongs to some other object, so
prevent the default __del__ implementation from closing that buffer.
"""The buffer this object contains belongs to some other object,
so prevent the default ``__del__`` implementation from closing
that buffer.
.. versionadded:: 8.3.2
"""
...
def fileno(self) -> int:
"""Return the file descriptor of the original stream, if one was
provided at construction time.
This allows C-level consumers (``faulthandler``, ``subprocess``,
signal handlers, ...) to obtain a valid fd without crashing, even
though the Python-level writes are redirected to an in-memory
buffer.
.. versionadded:: 8.3.3
"""
if self._original_fd >= 0:
return self._original_fd
return super().fileno()
@property
def name(self) -> str:
@@ -320,6 +355,20 @@ class CliRunner:
stream_mixer = StreamMixer()
# Preserve the original file descriptors so that C-level
# consumers (faulthandler, subprocess, etc.) can still obtain a
# valid fd from the redirected streams. The original streams
# may themselves lack a fileno() (e.g. when CliRunner is used
# inside pytest's capsys), so we fall back to -1.
def _safe_fileno(stream: t.IO[t.Any]) -> int:
try:
return stream.fileno()
except (AttributeError, io.UnsupportedOperation):
return -1
old_stdout_fd = _safe_fileno(old_stdout)
old_stderr_fd = _safe_fileno(old_stderr)
if self.echo_stdin:
bytes_input = echo_input = t.cast(
t.BinaryIO, EchoingStdin(bytes_input, stream_mixer.stdout)
@@ -335,7 +384,11 @@ class CliRunner:
text_input._CHUNK_SIZE = 1 # type: ignore
sys.stdout = _NamedTextIOWrapper(
stream_mixer.stdout, encoding=self.charset, name="<stdout>", mode="w"
stream_mixer.stdout,
encoding=self.charset,
name="<stdout>",
mode="w",
original_fd=old_stdout_fd,
)
sys.stderr = _NamedTextIOWrapper(
@@ -344,6 +397,7 @@ class CliRunner:
name="<stderr>",
mode="w",
errors="backslashreplace",
original_fd=old_stderr_fd,
)
@_pause_echo(echo_input) # type: ignore
@@ -390,12 +444,52 @@ class CliRunner:
old__getchar_func = termui._getchar
old_should_strip_ansi = utils.should_strip_ansi # type: ignore
old__compat_should_strip_ansi = _compat.should_strip_ansi
old_pdb_init = pdb.Pdb.__init__
termui.visible_prompt_func = visible_input
termui.hidden_prompt_func = hidden_input
termui._getchar = _getchar
utils.should_strip_ansi = should_strip_ansi # type: ignore
_compat.should_strip_ansi = should_strip_ansi
def _patched_pdb_init(
self: pdb.Pdb,
completekey: str = "tab",
stdin: t.IO[str] | None = None,
stdout: t.IO[str] | None = None,
**kwargs: t.Any,
) -> None:
"""Default ``pdb.Pdb`` to real terminal streams during
``CliRunner`` isolation.
Without this patch, ``pdb.Pdb.__init__`` inherits from
``cmd.Cmd`` which falls back to ``sys.stdin``/``sys.stdout``
when no explicit streams are provided. During isolation
those are ``BytesIO``-backed wrappers, so the debugger
reads from an empty buffer and writes to captured output,
making interactive debugging impossible.
By defaulting to ``sys.__stdin__``/``sys.__stdout__`` (the
original terminal streams Python preserves regardless of
redirection), debuggers can interact with the user while
``click.echo`` output is still captured normally.
This covers ``pdb.set_trace()``, ``breakpoint()``,
``pdb.post_mortem()``, and debuggers that subclass
``pdb.Pdb`` (ipdb, pdbpp). Explicit ``stdin``/``stdout``
arguments are honored and not overridden. Debuggers that
do not subclass ``pdb.Pdb`` (pudb, debugpy) are not
covered.
"""
if stdin is None:
stdin = sys.__stdin__
if stdout is None:
stdout = sys.__stdout__
old_pdb_init(
self, completekey=completekey, stdin=stdin, stdout=stdout, **kwargs
)
pdb.Pdb.__init__ = _patched_pdb_init # type: ignore[assignment]
old_env = {}
try:
for key, value in env.items():
@@ -426,6 +520,7 @@ class CliRunner:
utils.should_strip_ansi = old_should_strip_ansi # type: ignore
_compat.should_strip_ansi = old__compat_should_strip_ansi
formatting.FORCED_WIDTH = old_forced_width
pdb.Pdb.__init__ = old_pdb_init # type: ignore[method-assign]
def invoke(
self,

View File

@@ -57,7 +57,10 @@ def make_str(value: t.Any) -> str:
def make_default_short_help(help: str, max_length: int = 45) -> str:
"""Returns a condensed version of help string."""
"""Returns a condensed version of help string.
:meta private:
"""
# Consider only the first paragraph.
paragraph_end = help.find("\n\n")