From b4ea53e1ea82151b1ff3197f66a8d19a3887edb8 Mon Sep 17 00:00:00 2001 From: Stephen Rosen Date: Fri, 31 Jul 2026 18:20:59 -0500 Subject: [PATCH] Introduce register_log_level/reset_log_levels These two methods, housed in a dedicated module, call out to register custom log levels across various other components. Registering a log level does the following: - confirm the level and name are good - add the level and name to mappings - add a relevant named method to structlog-provided loggers - add the log level to level-based filtering bound loggers The reset method inverts these steps, removing from these same registered locations in reverse order. Names which are already used by `structlog` are forbidden, so no special checking or "undo rules" are needed for the reset. --- CHANGELOG.md | 4 + src/structlog/__init__.py | 3 + src/structlog/_custom_log_levels.py | 112 ++++++++++++++++++++++++++++ src/structlog/_native.py | 105 ++++++++++++++++---------- src/structlog/_output.py | 12 +++ tests/test_custom_log_levels.py | 99 ++++++++++++++++++++++++ 6 files changed, 294 insertions(+), 41 deletions(-) create mode 100644 src/structlog/_custom_log_levels.py create mode 100644 tests/test_custom_log_levels.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b23d71a7..2ce5336f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ You can find our backwards-compatibility policy [here](https://github.com/hynek/ ## [Unreleased](https://github.com/hynek/structlog/compare/26.1.0...HEAD) +### Added + +- `structlog.register_log_level()` and `structlog.reset_log_levels()` for registering and clearing custom log levels. + ## [26.1.0](https://github.com/hynek/structlog/compare/25.5.0...26.1.0) - 2026-06-06 diff --git a/src/structlog/__init__.py b/src/structlog/__init__.py index 0c72e360..feed8781 100644 --- a/src/structlog/__init__.py +++ b/src/structlog/__init__.py @@ -28,6 +28,7 @@ reset_defaults, wrap_logger, ) +from structlog._custom_log_levels import register_log_level, reset_log_levels from structlog._generic import BoundLogger from structlog._native import make_filtering_bound_logger from structlog._output import ( @@ -79,7 +80,9 @@ "is_configured", "make_filtering_bound_logger", "processors", + "register_log_level", "reset_defaults", + "reset_log_levels", "stdlib", "testing", "threadlocal", diff --git a/src/structlog/_custom_log_levels.py b/src/structlog/_custom_log_levels.py new file mode 100644 index 00000000..7fb6f08c --- /dev/null +++ b/src/structlog/_custom_log_levels.py @@ -0,0 +1,112 @@ +# SPDX-License-Identifier: MIT OR Apache-2.0 +# This file is dual licensed under the terms of the Apache License, Version +# 2.0, and the MIT License. See the LICENSE file in the root of this +# repository for complete details. + +from ._log_levels import LEVEL_TO_NAME, NAME_TO_LEVEL +from ._native import ( + add_new_custom_level_filtering, + remove_custom_level_filtering, +) +from ._output import add_new_custom_level_method, remove_custom_level_method + + +CUSTOM_LOG_LEVEL_NAMES: set[str] = set() +CUSTOM_LOG_LEVELS: set[int] = set() + +# Forbid use of well-known names that would cause confusion. +FORBIDDEN_LEVEL_NAMES: frozenset[str] = frozenset( + ( + # expected logger methods, sync and async variants + "critical", + "acritical", + "debug", + "adebug", + "err", + "aerr", + "error", + "aerror", + "exception", + "aexception", + "failure", + "afailure", + "fatal", + "afatal", + "info", + "ainfo", + "log", + "alog", + "warn", + "awarn", + "warning", + "awarning", + "msg", + "amsg", + # NOTSET is a special 0-valued level, but not a method name + "notset", + ) +) + + +def register_log_level(name: str, level: int, /) -> None: + """Register a custom log level for use with structlog. + + Loggers will be automatically enabled for use with this log level, with a method + named by the given level name. + + The log level will not be registered in the standard library `logging` module. + + Both the name and level must be distinct from names and levels already in use. + Names and levels which are already used will be rejected with an error, as will + level names which are not valid identifiers and level names which match structlog + method names. + + Args: + name: The name of the log level. + + level: + The numeric value of the log level as an integer. + This determines level-based filtering behavior. + """ + if not name.isidentifier(): + msg = f"Log level names must be valid identifiers. Got {name!r}" + raise ValueError(msg) + if name.startswith("_"): + msg = f"Log level names can't start with '_'. Got {name!r}" + raise ValueError(msg) + if name in FORBIDDEN_LEVEL_NAMES or name in NAME_TO_LEVEL: + msg = ( + "register_log_level() cannot be used with names which are " + f"already in use. Got: {name!r}" + ) + raise ValueError(msg) + if level in LEVEL_TO_NAME: + msg = ( + "register_log_level() cannot be used with levels which are " + f"already in use. Got {level!r}, " + f"which maps to {LEVEL_TO_NAME[level]!r}" + ) + raise ValueError(msg) + + CUSTOM_LOG_LEVEL_NAMES.add(name) + CUSTOM_LOG_LEVELS.add(level) + + NAME_TO_LEVEL[name] = level + LEVEL_TO_NAME[level] = name + add_new_custom_level_method(name) + add_new_custom_level_filtering(name) + + +def reset_log_levels() -> None: + """ + Reset all custom log levels and log level names. + """ + while CUSTOM_LOG_LEVEL_NAMES: + name = CUSTOM_LOG_LEVEL_NAMES.pop() + remove_custom_level_filtering(name) + remove_custom_level_method(name) + del NAME_TO_LEVEL[name] + + while CUSTOM_LOG_LEVELS: + level = CUSTOM_LOG_LEVELS.pop() + del LEVEL_TO_NAME[level] diff --git a/src/structlog/_native.py b/src/structlog/_native.py index 5775b8de..344fb944 100644 --- a/src/structlog/_native.py +++ b/src/structlog/_native.py @@ -152,56 +152,58 @@ def _maybe_interpolate(event: str, args: tuple[Any, ...]) -> str: return event % args -def _make_filtering_bound_logger(min_level: int) -> type[FilteringBoundLogger]: - """ - Create a new `FilteringBoundLogger` that only logs *min_level* or higher. +def _make_method( + min_level: int, + level: int, +) -> tuple[Callable[..., Any], Callable[..., Any]]: + if level < min_level: + return _nop, _anop - The logger is optimized such that log levels below *min_level* only consist - of a ``return None``. - """ + name = LEVEL_TO_NAME[level] - def make_method( - level: int, - ) -> tuple[Callable[..., Any], Callable[..., Any]]: - if level < min_level: - return _nop, _anop + def meth(self: Any, event: str, *args: Any, **kw: Any) -> Any: + return self._proxy_to_logger( + name, _maybe_interpolate(event, args), **kw + ) - name = LEVEL_TO_NAME[level] + async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any: + """ + .. versionchanged:: 23.3.0 + Callsite parameters are now also collected under asyncio. + """ + event = _maybe_interpolate(event, args) + + # Capture thread-specific info before handing off to the executor. + thread_token = _ASYNC_CALLING_THREAD.set( + (threading.get_ident(), threading.current_thread().name) + ) + scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] + ctx = contextvars.copy_context() - def meth(self: Any, event: str, *args: Any, **kw: Any) -> Any: - return self._proxy_to_logger( - name, _maybe_interpolate(event, args), **kw + try: + await asyncio.get_running_loop().run_in_executor( + None, + lambda: ctx.run( + lambda: self._proxy_to_logger(name, event, **kw) + ), ) + finally: + _ASYNC_CALLING_STACK.reset(scs_token) + _ASYNC_CALLING_THREAD.reset(thread_token) - async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any: - """ - .. versionchanged:: 23.3.0 - Callsite parameters are now also collected under asyncio. - """ - event = _maybe_interpolate(event, args) + meth.__name__ = name + ameth.__name__ = f"a{name}" - # Capture thread-specific info before handing off to the executor. - thread_token = _ASYNC_CALLING_THREAD.set( - (threading.get_ident(), threading.current_thread().name) - ) - scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] - ctx = contextvars.copy_context() + return meth, ameth - try: - await asyncio.get_running_loop().run_in_executor( - None, - lambda: ctx.run( - lambda: self._proxy_to_logger(name, event, **kw) - ), - ) - finally: - _ASYNC_CALLING_STACK.reset(scs_token) - _ASYNC_CALLING_THREAD.reset(thread_token) - meth.__name__ = name - ameth.__name__ = f"a{name}" +def _make_filtering_bound_logger(min_level: int) -> type[FilteringBoundLogger]: + """ + Create a new `FilteringBoundLogger` that only logs *min_level* or higher. - return meth, ameth + The logger is optimized such that log levels below *min_level* only consist + of a ``return None``. + """ def log(self: Any, level: int, event: str, *args: Any, **kw: Any) -> Any: if level < min_level: @@ -246,7 +248,7 @@ async def alog( meths: dict[str, Callable[..., Any]] = {"log": log, "alog": alog} for lvl, name in LEVEL_TO_NAME.items(): - meths[name], meths[f"a{name}"] = make_method(lvl) + meths[name], meths[f"a{name}"] = _make_method(min_level, lvl) meths["exception"] = exception meths["aexception"] = aexception @@ -284,3 +286,24 @@ async def alog( DEBUG: BoundLoggerFilteringAtDebug, NOTSET: BoundLoggerFilteringAtNotset, } + + +def add_new_custom_level_filtering(level_name: str, /) -> None: + level = NAME_TO_LEVEL[level_name] + for bound_logger in LEVEL_TO_FILTERING_LOGGER.values(): + meth, ameth = _make_method( + bound_logger.get_effective_level(None), # type: ignore[arg-type] + level, + ) + setattr(bound_logger, level_name, meth) + setattr(bound_logger, f"a{level_name}", ameth) + + LEVEL_TO_FILTERING_LOGGER[level] = _make_filtering_bound_logger(level) + + +def remove_custom_level_filtering(level_name: str, /) -> None: + level = NAME_TO_LEVEL[level_name] + for bound_logger in LEVEL_TO_FILTERING_LOGGER.values(): + delattr(bound_logger, level_name) + delattr(bound_logger, f"a{level_name}") + del LEVEL_TO_FILTERING_LOGGER[level] diff --git a/src/structlog/_output.py b/src/structlog/_output.py index 68fa976b..ca94b094 100644 --- a/src/structlog/_output.py +++ b/src/structlog/_output.py @@ -374,3 +374,15 @@ def __init__(self, file: BinaryIO | None = None): def __call__(self, *args: Any) -> BytesLogger: return BytesLogger(self._file, name=args[0] if args else None) + + +def add_new_custom_level_method(level_name: str, /) -> None: + """Register a new log level method.""" + for logger_class in (BytesLogger, PrintLogger, WriteLogger): + setattr(logger_class, level_name, logger_class.msg) + + +def remove_custom_level_method(level_name: str, /) -> None: + """Remove all registered log level methods.""" + for logger_class in (BytesLogger, PrintLogger, WriteLogger): + delattr(logger_class, level_name) diff --git a/tests/test_custom_log_levels.py b/tests/test_custom_log_levels.py new file mode 100644 index 00000000..149b1ded --- /dev/null +++ b/tests/test_custom_log_levels.py @@ -0,0 +1,99 @@ +# SPDX-License-Identifier: MIT OR Apache-2.0 +# This file is dual licensed under the terms of the Apache License, Version +# 2.0, and the MIT License. See the LICENSE file in the root of this +# repository for complete details. + +import logging +import re + +import pytest + +import structlog + + +@pytest.fixture(autouse=True) +def auto_reset_log_levels(): + yield + structlog.reset_log_levels() + + +def test_custom_log_levels_must_be_valid_identifiers(): + """Log levels have to be valid to use as method names.""" + with pytest.raises( + ValueError, match="Log level names must be valid identifiers" + ): + structlog.register_log_level("spam and eggs", 1) + + +def test_custom_log_levels_cant_be_private(): + """Private and dunder names are forbidden.""" + with pytest.raises( + ValueError, match="Log level names can't start with '_'" + ): + structlog.register_log_level("__init__", 1) + + +# we don't reproduce the whole list of forbidden names here, +# but at least some of the interesting parts +@pytest.mark.parametrize( + "levelname", ["notset", "msg", "amsg", "log", "alog", "info", "afatal"] +) +def test_custom_log_levels_cant_match_forbidden_list(levelname): + """Values in the special deny list are forbidden.""" + with pytest.raises( + ValueError, + match=re.escape( + r"register_log_level() cannot be used with names " + rf"which are already in use. Got: {levelname!r}" + ), + ): + structlog.register_log_level(levelname, 1) + + +def test_custom_log_levels_cant_match_existing_levels(): + """No duplicates by value.""" + with pytest.raises( + ValueError, + match=re.escape( + "register_log_level() cannot be used with levels which are " + f"already in use. Got {logging.INFO!r}, " + "which maps to 'info'" + ), + ): + structlog.register_log_level("detail", logging.INFO) + + +def test_custom_log_level_allows_logging_via_named_method(capsys): + """Logger objects are dynamically updated with registered levels.""" + print_logger = structlog.PrintLogger() + assert not hasattr(print_logger, "audit") + + structlog.register_log_level("audit", 100) + + assert hasattr(print_logger, "audit") + + print_logger.audit("hello") + out, err = capsys.readouterr() + assert "hello\n" == out + assert "" == err + + +def test_custom_log_level_can_be_filtered_out(cl): + """Filtering loggers can act on custom levels.""" + bl = structlog.make_filtering_bound_logger(logging.DEBUG)(cl, [], {}) + structlog.register_log_level("trace", logging.DEBUG - 1) + + bl.trace("ok") + assert [] == cl.calls + + bl.debug("yeah") + assert [("debug", (), {"event": "yeah"})] == cl.calls + + +def test_custom_log_level_can_be_allowed_by_filter(cl): + """Filtering loggers do not *always* filter out custom levels.""" + bl = structlog.make_filtering_bound_logger(logging.INFO)(cl, [], {}) + structlog.register_log_level("detail", logging.INFO + 1) + + bl.detail("ok") + assert [("detail", (), {"event": "ok"})] == cl.calls