Introduce register_log_level/reset_log_levels

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
 

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",

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]

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]

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)

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