# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Logging utilities.
```{todo}
Let the `-v`/`-q` counter reach beyond the current {class}`LogLevel` range, as
sketched by the `-vvvv` (trace) and `-q` (silence everything) notes that used
to live on `_VerbosityOption`:
- a `TRACE` pseudo-level below {attr}`LogLevel.DEBUG` (numeric value `5`,
mirroring `logging.DEBUG - 5`) so repeated `-v` can surface finer-grained
tracing past `DEBUG`;
- a `SILENT` pseudo-level above {attr}`LogLevel.CRITICAL` (any value above
`logging.CRITICAL`) so repeated `-q` can suppress every record, including
{attr}`LogLevel.CRITICAL`.
Both require extending {class}`LogLevel`, which ripples into the `--verbosity`
{class}`~click_extra.types.EnumChoice`, the {class}`Formatter` level-name color
lookup and the level-ordering tests. They are intentionally left out of the
symmetric-counter change that introduced `-q`, where the counter simply clamps
at `DEBUG`/`CRITICAL`.
```
"""
from __future__ import annotations
import logging
import sys
import weakref
from contextlib import nullcontext
from enum import IntEnum
from functools import partial
from gettext import gettext as _
from logging import (
FileHandler,
Handler,
Logger,
getLogger,
)
import click
from click._utils import UNSET
from click.types import IntRange
from . import context
from ._utils import patch_attr
from .parameters import ExtraOption, last_param
from .theme import get_current_theme
from .types import EnumChoice
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Generator, Iterable, Mapping, Sequence
from logging import LogRecord
from typing import IO, Any, ClassVar, Literal
logger = getLogger(__name__)
[docs]
class LogLevel(IntEnum):
"""Mapping of {ref}`canonical log level names <levels>` to their integer level.
That's our own version of [logging._nameToLevel](https://github.com/python/cpython/blob/3.14/Lib/logging/__init__.py#L115-L124),
but:
- sorted from lowest to highest verbosity,
- excludes the following levels:
- {data}`NOTSET <logging.NOTSET>`, which is considered internal
- `WARN`, which {meth}`is obsolete <logging.Logger.warning>`
- `FATAL`, which [shouldn't be used](https://github.com/python/cpython/issues/85013)
and has been [replaced by CRITICAL](https://github.com/python/cpython/blob/3.14/Lib/logging/__init__.py#L2150-L2154)
"""
CRITICAL = logging.CRITICAL
ERROR = logging.ERROR
WARNING = logging.WARNING
INFO = logging.INFO
DEBUG = logging.DEBUG
def __str__(self):
"""Use upper-case names as string representation."""
return self.name
DEFAULT_FORMAT = "{levelname}: {message}"
"""Format of a log record, written in the style of {data}`DEFAULT_STYLE`.
The default of {func}`basicConfig`, and the format of the handler an option of Click
Extra attaches to the `root` logger for the time of a run.
"""
DEFAULT_LEVEL: LogLevel = LogLevel.WARNING
"""{data}`WARNING <logging.WARNING>` is the default level we expect any loggers to starts their lives at.
{data}`WARNING <logging.WARNING>` has been chosen as it is [the level at which the default Python's global root logger is set up](https://github.com/python/cpython/blob/3.14/Lib/logging/__init__.py#L1997).
This value is also used as the default level for {class}`VerbosityOption` .
"""
DEFAULT_STYLE: Literal["%", "{", "$"] = "{"
"""Style of the format strings: ``{`` is the one of {meth}`str.format`.
The default of {func}`basicConfig`, where {func}`logging.basicConfig` defaults to `%`.
"""
_LEVEL_TOKEN: str = f"{context.META_NAMESPACE}_log_level_token"
"""Internal `ctx.meta` slot for the token a run owns its log levels with.
Set by the first verbosity option that applies a level, which also queues the
reset of the loggers. It lives in `ctx.meta` so the verbosity inheritance chain
(`_VerbosityOption` / `VerbosityOption` / `VerboseOption` / `QuietOption`)
queues that reset at most once per invocation, even when several verbosity
options are passed. See `_level_owner`.
"""
_COUNTER_SCANNED: str = f"{context.META_NAMESPACE}_verbosity_counter_scanned"
"""Internal sentinel marking that the `-v`/`-q` counter was pre-resolved.
The first verbosity option reaching {meth}`_VerbosityOption.handle_parse_result`
reads the raw `--verbose`/`--quiet` repetition counts from the parsed
`opts` and stashes them on the context before any option applies a level. This
way the option that fires first already reconciles the full counter and never
applies an intermediate, louder level that a later `-q` would have to lower
(which would leak its `Set ... DEBUG` debug trace).
"""
_CONFIGURED_HANDLERS: weakref.WeakSet[Handler] = weakref.WeakSet()
"""Handlers {func}`basicConfig` attached, so the ones of {func}`new_logger` too.
A logger holding one of them is set up the way its owner asked, whatever the
classes involved, and `_hold_invocation_handler` leaves it alone. The set holds
its handlers weakly, so it never keeps a discarded one alive.
"""
_HANDLER_TOKEN: str = f"{context.META_NAMESPACE}_log_handler_token"
"""Internal `ctx.meta` slot for the token a run holds the log handler with.
`ctx.meta` is shared down the context hierarchy, so the options of a command and
of its subcommands find the same token there, and a run takes one hold however
many of them it carries. See `_hold_invocation_handler`.
"""
_INVOCATION_HANDLERS: dict[str, Handler] = {}
"""The handler the options of Click Extra gave a logger for the runs in progress.
Keyed by logger name. Only the `root` logger gets one, and every run shares it:
a run started from within another one prints its records once.
A dict, for the two operations that keep it consistent when runs start and end
at the same time, in several threads or from a finalizer: `setdefault()`
installs a handler unless one is there, and `pop()` hands it to a single taker.
"""
_INVOCATION_HOLDERS: set[object] = set()
"""One token for each run holding the handler of `_INVOCATION_HANDLERS`.
The handler is detached when the last token is released. Counting the holders is
what keeps the handler in place for a run when the previous one failed while
parsing: the context of the failed run releases its token late, once it is
collected, which may happen in the middle of the next run.
"""
_level_owner: object | None = None
"""The token of the run that set the current log levels, until it gives them back.
Compared by `_take_saved_levels`. The context of a run that failed while parsing
is collected late, possibly in the middle of a later run, whose levels it must
not touch.
"""
_level_saved: dict[str, int] = {}
"""The level each logger had before a run changed it, by logger name.
Filled by the run of `_level_owner`, which gives these levels back when it ends.
A run that takes the levels over from another one keeps its entries: the levels
to give back are the ones found before the first of the two.
"""
[docs]
class StreamHandler(logging.StreamHandler):
"""A handler to output logs to the console.
Wraps {class}`logging.StreamHandler`, but use {func}`click.echo` to support color printing.
Only :py:data:`<stderr> <sys.stderr>` or :py:data:`<stdout> <sys.stdout>` are allowed as output stream.
If stream is not specified, :py:data:`<stderr> <sys.stderr>` is used by default
"""
_stderr_output: bool = True
"""{func}`click.echo`'s `err` parameter to be used at printing time."""
_stream: IO[Any] = sys.stderr
@property
def stream(self) -> IO[Any]:
"""The stream to which logs are written.
A proxy of the parent {class}`logging.StreamHandler`'s [stream attribute](https://github.com/python/cpython/blob/3.14/Lib/logging/__init__.py#L1129).
Redefined here to enforce checks on the stream value.
"""
return self._stream
@stream.setter
def stream(self, stream: IO[Any]) -> None:
if stream not in (sys.stderr, sys.stdout):
raise ValueError("Only <stderr> or <stdout> are allowed as output stream.")
self._stream = stream
# Sync click.echo()'s err parameter with the target stream.
self._stderr_output = stream == sys.stderr
[docs]
def emit(self, record: LogRecord) -> None:
"""Use {func}`click.echo` to print to the console.
Cooperates with any live terminal line currently drawing on the same
stream (a {class}`~click_extra.spinner.Spinner`, or an
{class}`~click_extra.spinner.OperationTrail` progress bar): the record
is then printed through its `echo`, which erases the in-progress render
first, so a log line emitted mid-draw lands on its own line instead of
garbling the indicator (and vice versa).
The color tri-state is resolved through
{func}`~click_extra.color.invocation_color` rather than left to
{func}`click.echo`'s own context lookup: a record emitted from a
background thread (a subprocess stream reader, a fan-out worker) has no
reachable Click context, and would otherwise ignore `--no-color` and
keep its ANSI codes on a TTY.
"""
try:
# Imported here, not at module level: spinner.py and color.py are
# leaves this logging module must not force-load (nor risk an import
# cycle with) just to emit a record.
from .color import invocation_color
from .spinner import _active_line
message = self.format(record)
color = invocation_color()
line = _active_line(self._stream)
if line is not None:
# The live line's echo writes raw: strip what click.echo would.
if color is False:
# Deferred import: keeps boltons off the startup path of every CLI.
from boltons.strutils import strip_ansi
message = strip_ansi(message)
line.echo(message)
else:
click.echo(message, err=self._stderr_output, color=color)
except RecursionError:
raise
# If exception occurs, dump the traceback to stderr.
except Exception: # noqa: BLE001
self.handleError(record)
class _InvocationHandler(StreamHandler):
"""The handler the options of Click Extra lend the `root` logger for a run.
It prints nothing once the last run let go of it. Code that saves the handlers
of a logger and puts them back later, as a fixture does to isolate a test, may
attach it again by then: each record of a later run would otherwise print a
second time, beside the handler of that run.
"""
def emit(self, record: LogRecord) -> None:
"""Print the record for as long as a run holds the handler."""
if self in _INVOCATION_HANDLERS.values():
super().emit(record)
[docs]
def basicConfig(
*,
# Arguments from Python's standard library's basicConfig:
filename: str | None = None,
filemode: str = "a",
format: str | None = DEFAULT_FORMAT,
datefmt: str | None = None,
style: Literal["%", "{", "$"] = DEFAULT_STYLE,
level: int | str | None = None,
stream: IO[Any] | None = None,
handlers: Iterable[Handler] | None = None,
force: bool = False,
encoding: str | None = None,
errors: str | None = "backslashreplace",
# New arguments specific to this function:
stream_handler_class: type[Handler] = StreamHandler,
file_handler_class: type[Handler] = FileHandler,
formatter_class: type[logging.Formatter] = Formatter,
) -> None:
"""Configure the global `root` logger.
This function is a wrapper around Python standard library's {func}`logging.basicConfig`,
but with additional parameters and tweaked defaults.
It sets up the global `root` logger, and optionally adds a file or stream handler to it.
Differences in default values:
========== =========================== ======================================
Argument {func}`basicConfig` default {func}`logging.basicConfig` default
========== =========================== ======================================
`style` ``{`` `%`
`format` ``{levelname}: {message}`` `%(levelname)s:%(name)s:%(message)s`
========== =========================== ======================================
This function takes the same parameters as {func}`logging.basicConfig`, but require them
to be all passed as explicit keywords arguments.
:param filename: Specifies that a {class}`logging.FileHandler` be created, using the
specified filename, rather than an :py:class:`StreamHandler`.
:param filemode: If *filename* is specified, open the file in this {func}`mode <.open>`.
Defaults to `a`.
:param format: Use the specified format string for the handler.
Defaults to ``{levelname}: {message}``.
:param datefmt: Use the specified date/time format, as accepted by
{func}`time.strftime`.
:param style: If format is specified, use this style for the format string:
- `%` for {ref}`printf-style <old-string-formatting>`,
- ``{`` for {meth}`str.format`,
- `$` for {class}`string.Template`.
Defaults to ``{``.
:param level: Set the `root` logger level to the specified {ref}`level <levels>`.
:param stream: Use the specified stream to initialize the
:py:class:`StreamHandler`. Note that this argument is incompatible with
*filename* - if both are present, a `ValueError` is raised.
:param handlers: If specified, this should be an iterable of already created
handlers to add to the `root` logger. Any handlers which don't already have a
formatter set will be assigned the default formatter created in this function.
Note that this argument is incompatible with *filename* or *stream* - if both
are present, a `ValueError` is raised.
:param force: If this argument is specified as `True`, any existing
handlers attached to the `root` logger are removed and closed, before carrying
out the configuration as specified by the other arguments.
:param encoding: {ref}`Name of the encoding <standard-encodings>` used to decode or
encode the file. To be specified along with *filename*, and passed to
{class}`logging.FileHandler` for opening the output file.
:param errors: Optional string that specifies {ref}`how encoding and decoding errors
are to be handled <error-handlers>` by the {class}`logging.FileHandler`. Defaults
to `backslashreplace`. Note that if `None` is specified, it will be passed as
such to {func}`open`.
```{important}
Always keep the signature of this function, the default values of its
parameters and its documentation in sync with the one from Python's standard
library.
```
These new arguments are available for better configurability:
:param stream_handler_class: A :py:class:`logging.Handler` class that will be used in
{func}`logging.basicConfig` to create a default stream-based handler. Defaults to
:py:class:`StreamHandler`.
:param file_handler_class: A :py:class:`logging.Handler` class that will be used in
{func}`logging.basicConfig` to create a default file-based handler. Defaults to
{class}`logging.FileHandler`.
:param formatter_class: A :py:class:`logging.Formatter` class of the formatter that
will be used in {func}`logging.basicConfig` to setup the default formatter. Defaults to
:py:class:`Formatter`.
```{note}
I don't like the camel-cased name of this function and would have called it
`basic_config()`, but it's kept this way for consistency with Python's
standard library {func}`logging.basicConfig`.
```
"""
# Only the arguments that are set are passed on: the standard library tests
# their presence, not their value.
kwargs: dict[str, Any] = {
name: value
for name, value in (
("filename", filename),
("filemode", filemode),
("format", format),
("datefmt", datefmt),
("style", style),
("level", level),
("stream", stream),
("handlers", handlers),
("force", force),
("encoding", encoding),
("errors", errors),
("stream_handler_class", stream_handler_class),
("file_handler_class", file_handler_class),
("formatter_class", formatter_class),
)
if value is not None
}
# The standard library reads filemode for the handler it creates itself, and
# rejects it as unrecognised beside handlers of the caller.
if handlers is not None:
kwargs.pop("filemode", None)
call_str = ", ".join(f"{k}={v!r}" for k, v in kwargs.items())
logger.debug(f"Call basicConfig({call_str})")
# The standard library sets handlers up when told to replace the existing ones,
# or when there is none. Every handler found afterwards is then one it attached.
configures = force or not logging.root.handlers
# Consume along the way each kwargs' parameter not recognized by basicConfig.
with (
patch_attr(logging, "StreamHandler", kwargs.pop("stream_handler_class")),
patch_attr(logging, "FileHandler", kwargs.pop("file_handler_class")),
patch_attr(logging, "Formatter", kwargs.pop("formatter_class")),
):
logging.basicConfig(**kwargs)
if configures:
_CONFIGURED_HANDLERS.update(logging.root.handlers)
[docs]
def new_logger(
name: str = logging.root.name,
*,
propagate: bool = False,
force: bool = True,
**kwargs,
) -> Logger:
"""Setup a logger in the style of Click Extra.
By default, this helper will:
- {func}`Fetch the logger <logging.getLogger>` registered under the `name` parameter, or creates a new one
with that name if it doesn't exist,
- Set the logger's {attr}`propagate <logging.Logger.propagate>` attribute to `False`,
- Force removal of any existing handlers and formatters attached to the logger,
- Attach a new :py:class:`StreamHandler` with :py:class:`Formatter`,
- Return the logger object.
This function is a wrapper around {func}`basicConfig` and takes the same keywords arguments.
```{warning}
The removal reaches every handler of the logger, not only the ones Click Extra
attached. Left on its default `name`, this function sets up the `root`
logger, which the whole process shares: the handlers an application or a test
runner attached to it are removed and closed.
Call it from the entry point of a program instead of at import time, or name
a logger of your own. With `force=False` the handlers stay, and the function
sets nothing up on a logger that already has one.
```
```{note}
The options of Click Extra do not call this function for the `root` logger.
They attach their handler beside the others, for the time of a run. See
{meth}`~click_extra.parameters.ExtraOption.add_to_parser`.
```
:param name: ID of the logger to setup. Defaults to the name of Python's `root`
logger. If a logger with the provided name is not found in the global
registry, a new logger with that name will be created.
:param propagate: Sets the logger's {attr}`propagate <logging.Logger.propagate>` attribute. Defaults to `False`.
:param force: Same as the *force* parameter from {func}`logging.basicConfig` and {func}`basicConfig`. Defaults to `True`.
:param kwargs: Any other keyword parameters supported by {func}`logging.basicConfig` and {func}`basicConfig`.
"""
if name == logging.root.name:
logger = logging.root
root_logger_patch = nullcontext()
else:
logger = getLogger(name) # type: ignore[assignment]
logger.propagate = propagate
root_logger_patch = patch_attr( # type: ignore[assignment]
logging,
"root",
logger,
)
with root_logger_patch:
basicConfig(force=force, **kwargs)
return logger
def _hold_invocation_handler(ctx: click.Context) -> None:
"""Give the `root` logger Click Extra's handler for the run `ctx` belongs to.
The handler goes beside the ones already attached, which stay in place. The
`root` logger belongs to the whole process: an application that embeds the
CLI, or a test runner that captures log records, has its own handlers there.
Nothing is attached when {func}`basicConfig` or {func}`new_logger` set the
`root` logger up already: the handlers and the format chosen there are kept.
One handler serves every run in progress, and each run holds it with a token
of its own: the subcommands of a run share its token, and a run started from
within another one adds a second. A record so prints once, and the handler
stays for as long as one run needs it.
A run releases its token when its root context closes, and through
`click_extra.context._queue_run_cleanup` when it fails while parsing.
"""
if context.get(ctx, _HANDLER_TOKEN) in _INVOCATION_HOLDERS:
return
root = getLogger()
if any(handler in _CONFIGURED_HANDLERS for handler in tuple(root.handlers)):
return
token = object()
# Hold first: from here on, no release can take the handler away.
_INVOCATION_HOLDERS.add(token)
handler = _INVOCATION_HANDLERS.get(root.name)
if handler is None:
handler = _InvocationHandler()
handler.setFormatter(Formatter(fmt=DEFAULT_FORMAT, style=DEFAULT_STYLE))
# Another run may have installed one in the meantime: that one is used.
handler = _INVOCATION_HANDLERS.setdefault(root.name, handler)
# Take off the handler of an earlier run, which somebody put back after that
# run let go of it.
for other in tuple(root.handlers):
if isinstance(other, _InvocationHandler) and other is not handler:
root.removeHandler(other)
# A no-op when a run in progress attached it. A handler somebody else took off
# the logger in the meantime is put back.
root.addHandler(handler)
context.set(ctx, _HANDLER_TOKEN, token)
release = partial(_release_invocation_handler, token)
root_ctx = ctx.find_root()
root_ctx.call_on_close(release)
context._queue_run_cleanup(root_ctx, release)
def _release_invocation_handler(token: object) -> None:
"""Release the hold of a run, and detach the handler once no run holds it.
Safe to call more than once with the same token: the close callback, the
finalizer and the `_scoped_run` block of the command may each release the
same run.
A finalizer runs where Python collects the context, so in any thread and in
the middle of any call, the start of another run included. No lock guards
against that: a finalizer waiting for a lock its own thread holds never gets
it. Each step below is a single operation of a set or of a dict instead, in
an order that leaves a run starting at that moment with its handler.
"""
_INVOCATION_HOLDERS.discard(token)
if _INVOCATION_HOLDERS:
return
root = getLogger()
handler = _INVOCATION_HANDLERS.pop(root.name, None)
if handler is None:
return
# A run that started since the check above may count on this handler. It is
# put back, unless that run installed one of its own.
if (
_INVOCATION_HOLDERS
and _INVOCATION_HANDLERS.setdefault(root.name, handler) is handler
):
return
root.removeHandler(handler)
handler.close()
def _take_saved_levels(token: object) -> dict[str, int] | None:
"""Hand the levels to give back over to the run that owns them.
`None` for any other run: one whose levels a later run took over, or one that
gave them back already. The owner so gets them once, however many of its
cleanups ask.
"""
global _level_owner, _level_saved
if _level_owner is not token:
return None
_level_owner = None
saved, _level_saved = _level_saved, {}
return saved
def _expire_levels(token: object) -> None:
"""Give the loggers of a run their levels back, if that run still owns them.
Called for a run whose context Click never closed, so whose
{meth}`_VerbosityOption.reset_loggers` never ran.
Silent, unlike that method: the call may come from a finalizer, in the middle
of a later run, and a debug record emitted there would print in its output.
"""
for name, level in (_take_saved_levels(token) or {}).items():
getLogger(name).setLevel(level)
def _resulting_level(managed_logger: Logger, level: int) -> str:
"""Name the level `managed_logger` works at once its own is set to `level`.
A logger given back {data}`NOTSET <logging.NOTSET>` follows its parent again,
as the representation of a logger shows it: `<Logger almanac (WARNING)>`.
"""
if not level and managed_logger.parent:
level = managed_logger.parent.getEffectiveLevel()
return logging.getLevelName(level)
def _fold_level(
base: LogLevel,
*,
verbosity: LogLevel,
verbose: int,
quiet: int,
debug: bool,
) -> LogLevel:
"""Fold the raw verbosity requests into one level.
See {meth}`_VerbosityOption.resolve_level` for the rule.
"""
# `--debug` asks for the loudest level there is, so no counter and no
# `--verbosity` can outrank it. Answered here rather than by writing
# VERBOSITY, which would make the winner depend on which option Click
# happened to process last.
if debug:
return LogLevel.DEBUG
net = verbose - quiet
if net == 0:
return verbosity
levels = tuple(LogLevel)
# Higher index == more verbose. `-v` raises the index, `-q` lowers it.
counter_index = min(max(levels.index(base) + net, 0), len(levels) - 1)
counter = levels[counter_index]
# `min` picks the more verbose (lowest numeric) level, `max` the quieter.
return min(counter, verbosity) if net > 0 else max(counter, verbosity)
class _VerbosityOption(ExtraOption):
"""A base class implementing all the common helpers to manipulated logger's
verbosity.
Sets the level of the provided logger. If no logger is provided, sets the level of
the global `root` logger.
```{important}
The internal `click_extra` logger will be aligned to the level set through
this class.
```
```{caution}
This class is not intended to be used as-is. It is the internal place where
the level requested by the three competing options is reconciled:
- `--verbosity LEVEL` sets an absolute level (defaults to
{const}`DEFAULT_LEVEL`).
- `--verbose`/`-v` raises the verbosity, one {class}`LogLevel` step per
repetition.
- `--quiet`/`-q` lowers the verbosity, one {class}`LogLevel` step per
repetition.
- `--debug` is `--verbosity DEBUG` under the name people reach for, and
outranks all three.
`-v` and `-q` form a single signed counter around the base level:
`net = (number of -v) - (number of -q)`. The counter is clamped to the
{class}`LogLevel` range, so it never reaches past {attr}`LogLevel.DEBUG`
(loudest) nor {attr}`LogLevel.CRITICAL` (quietest). See
{meth}`resolve_level` for the way the counter is reconciled with an explicit
`--verbosity`.
```
```{note}
`-q` only lowers the *logging* verbosity. It deliberately does not silence
{func}`click.echo`: a command's primary output is not a diagnostic and stays
on its stream.
```
"""
logger_name: str
"""The ID of the logger to set the level to.
This will be provided to {func}`logging.getLogger` to fetch the logger object, and
as such, can be a dot-separated string to build hierarchical loggers.
"""
_help_verb: str | None = None
"""Direction word used by {meth}`get_help_record` to build the default help.
Set to `Increase` by {class}`VerboseOption` and `Decrease` by
{class}`QuietOption`. Left `None` on the base and on {class}`VerbosityOption`,
which both render their help verbatim instead of deriving it from the base level.
"""
def get_help_record(self, ctx: click.Context) -> tuple[str, str] | None:
"""Dynamically generates the default help message.
We need that patch because {meth}`get_base_level` depends on the context, so we
cannot hard-code the help message as the option's `__init__` default.
Only the `-v`/`-q` counting options derive their help from the base level
(via {attr}`_help_verb`). The base and {class}`VerbosityOption` carry a static
help and fall straight through to the parent implementation.
"""
help_message_patch = nullcontext()
if self.help is None and self._help_verb is not None:
help_message_patch = patch_attr( # type:ignore[assignment]
self,
"help",
(
f"{self._help_verb} the default {self.get_base_level(ctx)} "
"verbosity by one level for each additional repetition of the "
"option."
),
)
with help_message_patch:
return super().get_help_record(ctx)
@property
def all_loggers(self) -> Generator[Logger, None, None]:
"""Returns the list of logger IDs affected by the verbosity option.
Will returns `click_extra` internal logger first, then the option's
{attr}`logger_name`.
"""
for name in ("click_extra", self.logger_name):
yield getLogger(name)
def reset_loggers(self, levels: Mapping[str, int] | None = None) -> None:
"""Gives all loggers managed by the option their level back.
At the end of a run, that is the level each logger had before the run: an
application that set its own level on the `root` logger finds it again.
```{important}
Loggers are reset in reverse order to ensure the internal logger is changed
last. That way the internal `click_extra` logger can report its ongoing
logger-altering operations while using the logging facilities itself.
```
```{danger}
Resetting loggers is extremely important for unittests. Because they're
global, loggers have tendency to leak and pollute their state between
multiple test calls.
```
:param levels: The level to give back to each logger, by logger name. A
logger it does not name is reset to {const}`DEFAULT_LEVEL`, and so is
every logger when it is not provided.
"""
levels = levels or {}
for managed_logger in list(self.all_loggers)[::-1]:
level = levels.get(managed_logger.name, DEFAULT_LEVEL.value)
target = _resulting_level(managed_logger, level)
logger.debug(f"Reset {managed_logger} to {target}.")
managed_logger.setLevel(level)
def _reset_on_close(self, token: object) -> None:
"""Give the loggers their levels back when the context of the run closes.
The close callback of a run that set levels. Does nothing for a run whose
levels a later run took over, and gave back. See `_expire_levels` for a
run whose context never closes.
"""
saved = _take_saved_levels(token)
if saved is None:
return
self.reset_loggers(saved)
# The loggers of an earlier run that failed while parsing, whose levels
# this run took over: none of them is one this option reports on.
managed = {managed_logger.name for managed_logger in self.all_loggers}
for name, level in saved.items():
if name not in managed:
getLogger(name).setLevel(level)
def handle_parse_result(
self,
ctx: click.Context,
opts: Mapping[str, Any],
args: list[str],
) -> tuple[Any, list[str]]:
"""Pre-resolve the `-v`/`-q` counter before any level is applied.
The first verbosity option to be processed reads the raw `--verbose` and
`--quiet` repetition counts straight from the parsed `opts` and stashes them
on the context. By the time any verbosity callback applies a level,
{meth}`resolve_level` already sees the full counter, so the option that fires
first lands on the final level directly instead of applying a louder
intermediate that a later `-q` would have to walk back. See
{data}`_COUNTER_SCANNED`.
"""
if not ctx.resilient_parsing and not context.get(ctx, _COUNTER_SCANNED):
context.set(ctx, _COUNTER_SCANNED, True)
for klass, key in (
(VerboseOption, context.VERBOSE),
(QuietOption, context.QUIET),
):
option = last_param(ctx.command.params, klass)
if option is not None:
value, _ = option.consume_value(ctx, opts)
context.set(ctx, key, value)
return super().handle_parse_result(ctx, opts, args)
def get_base_level(self, ctx: click.Context) -> LogLevel:
"""Returns the base level the `-v`/`-q` counter is anchored at.
Sourced from the {attr}`~VerbosityOption.default` of any
{class}`VerbosityOption` declared on the current command. When the
`-v`/`-q` options are used standalone (no `--verbosity`), it defaults to
{const}`DEFAULT_LEVEL`.
"""
verbosity_option = last_param(ctx.command.params, VerbosityOption)
if verbosity_option is None:
return DEFAULT_LEVEL
return verbosity_option.default # type: ignore[return-value]
def resolve_level(self, ctx: click.Context) -> LogLevel:
"""Reconcile `--verbosity`, `-v` and `-q` into a single level.
Reads the raw value each option recorded on the context
({data}`~click_extra.context.VERBOSITY`,
{data}`~click_extra.context.VERBOSE` and
{data}`~click_extra.context.QUIET`) and folds them with this rule:
- `--debug` wins outright, naming the loudest level there is;
- `net = verbose - quiet`;
- `net == 0`: the `--verbosity` value wins (its default when the user did
not pass it);
- `net > 0`: the more verbose of the counter result and the `--verbosity`
value wins;
- `net < 0`: the more quiet of the counter result and the `--verbosity`
value wins.
The counter starts from {meth}`get_base_level` and is clamped to the
{class}`LogLevel` range. This keeps `-v` backward compatible (it still counts
up from the default and the loudest request wins), while letting `-q` mirror
it downwards.
"""
base = self.get_base_level(ctx)
return _fold_level(
base,
verbosity=context.get(ctx, context.VERBOSITY, base),
verbose=context.get(ctx, context.VERBOSE, 0),
quiet=context.get(ctx, context.QUIET, 0),
debug=context.get(ctx, context.DEBUG, False),
)
def apply_verbosity(self, ctx: click.Context) -> None:
"""Reconcile the requested verbosity and apply it to all managed loggers.
Called by every verbosity option after it has recorded its own raw value, so
the last option to fire reconciles the complete picture via
{meth}`resolve_level`. The reconciled level is published on `ctx.meta` under
{data}`~click_extra.context.VERBOSITY_LEVEL` and only (re)applied when it
actually changes, to keep the debug trace free of redundant `Set` lines.
The loggers get the level they had before the run back when it ends: through
{meth}`reset_loggers` when its context closes, and through `_expire_levels`
when it fails while parsing, which leaves the context never closed.
```{note}
A run started from within another one gives back the levels found before
the outer run, not the ones the outer run set: the outer run works at
those for the rest of its time.
```
"""
global _level_owner
# Skip logger reconfiguration during help rendering, shell completion,
# and any `make_context(resilient_parsing=True)` path.
if ctx.resilient_parsing:
return
new_level = self.resolve_level(ctx)
# Idempotent: several verbosity options flow through here, so without this
# guard the same level would be re-applied and re-logged once per option.
if context.get(ctx, context.VERBOSITY_LEVEL) == new_level:
return
context.set(ctx, context.VERBOSITY_LEVEL, new_level)
# Own the levels before setting them. An earlier run that failed while
# parsing, if it gives its levels back now, then leaves these alone.
token = context.get(ctx, _LEVEL_TOKEN)
queued = token is not None
if not queued:
token = object()
context.set(ctx, _LEVEL_TOKEN, token)
_level_owner = token
for managed_logger in self.all_loggers:
# Keep the level to give back, unless a run before this one did: the
# level found now is then one that run set.
_level_saved.setdefault(managed_logger.name, managed_logger.level)
managed_logger.setLevel(new_level.value)
logger.debug(f"Set {managed_logger} to {new_level}.")
# Queue the reset at most once per ctx. All verbosity options flow through
# this method, so without a guard the same reset would be queued more than
# once on Context._close_callbacks.
if not queued:
ctx.call_on_close(partial(self._reset_on_close, token))
context._queue_run_cleanup(ctx, partial(_expire_levels, token))
def set_level(self, ctx: click.Context, param: click.Parameter, value: Any) -> None:
"""Base callback: subclasses record their raw value first, then reconcile.
The base implementation only triggers reconciliation. Each subclass overrides
it to stash its own raw value on the context before delegating here.
"""
self.apply_verbosity(ctx)
def __init__(
self,
param_decls: Sequence[str] | None = None,
*,
default_logger: Logger | str = logging.root.name,
expose_value: bool = False,
is_eager: bool = True,
**kwargs: Any,
) -> None:
"""Set up a verbosity-altering option.
:param default_logger: The logger the option sets the level of:
- A {class}`logging.Logger` object is used as it is.
- The name of the `root` logger, which is the default, selects it.
- A name found in the global registry selects that logger as it is.
- Any other name creates the logger with {func}`new_logger`.
Building the option never touches the `root` logger, which the whole
process shares. Whichever logger is selected, the `root` one gets Click
Extra's handler for the time of each run of the command: see
{meth}`~click_extra.parameters.ExtraOption.add_to_parser`.
"""
# A logger object has been provided, fetch its name.
if isinstance(default_logger, Logger):
self.logger_name = default_logger.name
# The root logger is never found in Logger.manager.loggerDict.
elif default_logger in (None, logging.root.name):
self.logger_name = logging.root.name
# Use the provided string if it is found in the registry.
elif default_logger in Logger.manager.loggerDict:
self.logger_name = default_logger
# Create a new logger with Click Extra's default configuration.
else:
logger = new_logger(name=default_logger)
self.logger_name = logger.name
kwargs.setdefault("callback", self.set_level)
super().__init__(
param_decls=param_decls,
expose_value=expose_value,
is_eager=is_eager,
**kwargs,
)
[docs]
class VerbosityOption(_VerbosityOption):
"""`--verbosity LEVEL` option to set the log level of `_VerbosityOption`."""
[docs]
def set_level(
self, ctx: click.Context, param: click.Parameter, value: LogLevel
) -> None:
"""Record the `--verbosity` value, then reconcile.
The value passed to `--verbosity` is saved in
`ctx.meta[click_extra.context.VERBOSITY]`.
"""
context.set(ctx, context.VERBOSITY, value)
self.apply_verbosity(ctx)
def __init__(
self,
param_decls: Sequence[str] | None = None,
*,
default_logger: Logger | str = logging.root.name,
default: LogLevel = DEFAULT_LEVEL,
metavar: str = "LEVEL",
type: click.ParamType | Any = EnumChoice(LogLevel),
help: str = _("Either {log_levels}.").format(
log_levels=", ".join(map(str, LogLevel))
),
**kwargs: Any,
) -> None:
if not param_decls:
param_decls = ("--verbosity",)
super().__init__(
param_decls=param_decls,
default_logger=default_logger,
default=default,
metavar=metavar,
type=type,
help=help,
**kwargs,
)
[docs]
class DebugOption(_VerbosityOption):
"""`--debug` flag, a shorthand for `--verbosity DEBUG`.
The one level people actually reach for by name, so it earns a flag of its
own. The quiet end of the scale needs no such thing: `-q` already walks
down it, repeatably, and nobody types `--critical` to be told less.
```{note}
It outranks `--verbosity`, `-v` and `-q` alike rather than joining their
arithmetic, because it asks for the loudest level there is and no
combination of the other three can ask for more. See
{meth}`_VerbosityOption.resolve_level`.
```
"""
[docs]
def set_level(
self, ctx: click.Context, param: click.Parameter, value: bool
) -> None:
"""Record that `--debug` was passed, then reconcile.
The flag is saved in `ctx.meta[click_extra.context.DEBUG]`, so
downstream code can tell a level that came from `--debug` from the same
level spelled `--verbosity DEBUG`.
"""
if value:
context.set(ctx, context.DEBUG, True)
self.apply_verbosity(ctx)
def __init__(
self,
param_decls: Sequence[str] | None = None,
*,
default_logger: Logger | str = logging.root.name,
is_flag: bool = True,
default: bool = False,
help: str = _("Shorthand for --verbosity {level}.").format(
level=LogLevel.DEBUG
),
**kwargs: Any,
) -> None:
if not param_decls:
param_decls = ("--debug",)
super().__init__(
param_decls=param_decls,
default_logger=default_logger,
is_flag=is_flag,
default=default,
help=help,
**kwargs,
)
class _CounterOption(_VerbosityOption):
"""Shared machinery of the `-v`/`-q` counting twins.
Each subclass declares the `ctx.meta` key its raw repetition count is
recorded under ({attr}`_counter_key`) and its direction word
({attr}`_help_verb`, also reused by the debug trace); the recording,
reconciliation and tracing below are common to both.
"""
_counter_key: ClassVar[str]
"""The `ctx.meta` key the raw repetition count is recorded under."""
def set_level(self, ctx: click.Context, param: click.Parameter, value: int) -> None:
"""Record the repetition count, then reconcile.
The number of repetitions is saved on the context under
{attr}`_counter_key` and folded into the verbosity counter by
`_VerbosityOption.resolve_level`.
"""
context.set(ctx, self._counter_key, value)
self.apply_verbosity(ctx)
# Report the net effect after the level has been applied, so the message has a
# chance to be seen at DEBUG level.
if value and not ctx.resilient_parsing:
logger.debug(
f"{self._help_verb}d log verbosity by {value} levels: "
f"from {self.get_base_level(ctx)} "
f"to {context.get(ctx, context.VERBOSITY_LEVEL)}."
)
def __init__(
self,
param_decls: Sequence[str] | None = None,
*,
count: bool = True,
**kwargs: Any,
) -> None:
# Force type and default to have them aligned with the counting option's
# original behavior:
# https://github.com/pallets/click/blob/5dd6288/src/click/core.py#L2612-L2618
kwargs["type"] = IntRange(min=0)
kwargs["default"] = 0
super().__init__(
param_decls=param_decls,
count=count,
**kwargs,
)
[docs]
class VerboseOption(_CounterOption):
"""`--verbose`/`-v` option to raise the log level of `_VerbosityOption` by
one step per repetition.
Each `-v` raises the verbosity by one {class}`LogLevel` step. The option can be
repeated, so `-vv` (or `-v -v`) raises it by two steps.
The base level the counter starts from is sourced from
`VerbosityOption.default`. So with `--verbosity`'s default left at
`WARNING`:
- `-v` raises the level to `INFO`,
- `-vv` raises the level to `DEBUG`,
- any further repetition is clamped at the loudest level, so `-vvvvv` for example
resolves to `DEBUG`.
`-v` shares a single signed counter with {class}`QuietOption`'s `-q`, so the two
cancel out: `-v -q` leaves the level unchanged. See
`_VerbosityOption.resolve_level` for the full reconciliation rule with
`--verbosity`.
"""
_help_verb = "Increase"
_counter_key = context.VERBOSE
def __init__(
self,
param_decls: Sequence[str] | None = None,
**kwargs: Any,
) -> None:
if not param_decls:
param_decls = ("--verbose", "-v")
super().__init__(param_decls=param_decls, **kwargs)
[docs]
class QuietOption(_CounterOption):
"""`--quiet`/`-q` option to lower the log level of `_VerbosityOption` by
one step per repetition.
The symmetric counterpart of {class}`VerboseOption`: where `-v` raises the
verbosity one {class}`LogLevel` step at a time, `-q` lowers it. Starting from
`VerbosityOption.default` (`WARNING` by default):
- `-q` lowers the level to `ERROR`,
- `-qq` lowers the level to `CRITICAL`,
- any further repetition is clamped at the quietest level, so `-qqqqq` for example
resolves to `CRITICAL`.
`-q` shares a single signed counter with {class}`VerboseOption`'s `-v`, so the
two cancel out: `-v -q` leaves the level unchanged. See
`_VerbosityOption.resolve_level` for the full reconciliation rule with
`--verbosity`.
"""
_help_verb = "Decrease"
_counter_key = context.QUIET
def __init__(
self,
param_decls: Sequence[str] | None = None,
**kwargs: Any,
) -> None:
if not param_decls:
param_decls = ("--quiet", "-q")
super().__init__(param_decls=param_decls, **kwargs)
[docs]
def requested_level(
ctx: click.Context,
opts: Mapping[str, Any],
) -> LogLevel | None:
"""The level the verbosity options of `ctx.command` will settle on.
Read from the parsed `opts`, the environment and the defaults, without running
any of the options' callbacks, so an eager option processed before them can
honor it. {class}`~click_extra.config.ConfigOption` relies on it: it loads the
configuration before any verbosity option applies a level, yet `-q` and a
quieter `--verbosity` still silence its status lines.
Returns `None` when the command carries no verbosity option, or when a value
does not convert: the regular parse reports that error in its turn.
"""
options: dict[type[_VerbosityOption], _VerbosityOption | None] = {
klass: last_param(ctx.command.params, klass)
for klass in (VerbosityOption, VerboseOption, QuietOption, DebugOption)
}
present = [option for option in options.values() if option is not None]
if not present:
return None
def read(klass: type[_VerbosityOption], fallback: Any) -> Any:
option = options[klass]
if option is None:
return fallback
value, _ = option.consume_value(ctx, opts)
if value is None or value is UNSET:
return fallback
return option.type_cast_value(ctx, value)
base = present[0].get_base_level(ctx)
try:
return _fold_level(
base,
verbosity=read(VerbosityOption, base),
verbose=read(VerboseOption, 0),
quiet=read(QuietOption, 0),
debug=read(DebugOption, False),
)
except click.ClickException:
return None