Skip to content
Conduit

skill

lib-log-rich

bitranox/lib_log_richcommunitylistedunreviewed
Use skill

Set up

Skills are installed by placing their folder where your agent loads skills.

Download ZIP from GitHub Open the repository

The download comes from github.com/bitranox/lib_log_rich itself, not from our server, so what you get is exactly what its author published.

Install instructions come from the README

Skills do not use an MCP config. Follow the steps in the README below.

README

synced Oct 25, 2026

lib_log_rich

CI CodeQL License: MIT Open in Codespaces PyPI PyPI - Downloads Code Style: Ruff codecov Maintainability

  • Enjoy vibrant Rich-style colored console output with clean formatting.
  • Tailor colored (or monochrome) logs with flexible templates and themes, while your OS takes care of persistence.
  • Forget about manual logfile rotation.
  • Export logs as JSON, HTML or plain text whenever you need.
  • Send events to journald, Windows Event Log, or Graylog with ease.
  • Get optional OpenTelemetry integration for deeper observability.
  • Configure logging per app/user/host via configfile or environment
import logging
import lib_log_rich
from lib_log_rich.runtime import attach_std_logging

lib_log_rich.init(lib_log_rich.RuntimeConfig(service="my-app", environment="dev"))
attach_std_logging()

logging.info("Ready to go!")  # existing logging.* calls to the root logger just work

logger = logging.getLogger(__name__)  # log to any custom logger
logger.info("Also works fine!")  # logging in third party libraries just work

lib_log_rich.shutdown()  # shutdown to make sure all records are written to backends like graylog

Table of Contents

Integration Decision Guide

  • Prefer lib_log_rich.getLogger(...) and LoggerProxy when you own the emitting code. You gain the full structured pipeline: Rich console output, journald/Windows Event Log/Graylog fan-out, payload sanitisation, context binding, queue-based delivery, and diagnostic return values without touching stdlib internals.
  • Attach the StdlibLoggingHandler via attach_std_logging() when you must support existing logging hierarchies (framework defaults, third-party libraries, legacy modules). The handler normalises LogRecord objects into the same runtime, preserves extra payloads, copies call-site metadata (pathname, lineno, funcName), honours exc_info/stack_info, and skips recursion for events originating inside lib_log_rich.
  • Mixing both approaches is common: initialise the runtime once, use LoggerProxy in new code, and bridge whatever still relies on the standard library. All routes share the same ring buffer, payload limits, diagnostics, CLI tooling, and dump facilities.

Getting Started

Overview

Rich-powered logging backbone with contextual metadata, multi-target fan-out (console, journald, Windows Event Log, Graylog), ring-buffer dumps, and queue-based decoupling for multi-threaded workloads.
Rich renders multi-colour output tuned to each terminal, while adapters and dump exporters support configurable formats and templates.
Each runtime captures the active user, short hostname, process id, and PID chain automatically, so every sink receives consistent system identity fields.
The public API stays intentionally small: initialise once, bind context, emit logs (with per-event extra payloads), dump history in text/JSON/HTML, and shut down cleanly.

Python requirement: lib_log_rich targets Python 3.10 and newer. Free-threaded builds: 3.14t and 3.15t are supported, 3.13t is not (see JSON encoder and free-threaded Python). Core dependencies: lib_cli_exit_tools, orjson (below Python 3.14; from 3.14 the optional orjson extra), pydantic, rich, rich-click, and python-dotenv (the version floors live in pyproject.toml).

  • colored terminal logs via rich, with UTC or local timestamps
  • supports journald
  • supports Windows Event Logs
  • supports Graylog via Gelf (and gRPC after adding Open Telemetry Support)
  • supports quick log-dump with filtering from the ringbuffer without leaving the application
  • runtime configuration validated via Pydantic models, yielding structured errors and JSON schemas
  • per-event payload guards (4KB messages, 8KB extras, depth limits) configurable via payload_limits
  • independent log levels: console_level, backend_level, and graylog_level are completely independent - each gates events to its respective adapter (console, journald/EventLog, Graylog) without affecting the others. Use get_minimum_log_level() to find the lowest threshold for stdlib integration.
  • opt-in .env loading (same precedence for CLI and programmatic use)
  • Open Telemetry Support on user (Your) request - not implemented yet (because I do not need it myself). If You need it, let me know.
  • optional diagnostic_hook callback that observes the runtime without modifying it. The hook reports queue activity (drops, worker failures, drop-handler errors), rate limiting, and payload clamps so you can surface metrics, alerts, or dashboards while keeping the logging pipeline decoupled from any monitoring stack.
  • queue-backed console adapters (ANSI or HTML) - QueueConsoleAdapter for threads and AsyncQueueConsoleAdapter for asyncio - exposed via console_adapter_factory, handy for Textual panes, SSE/WebSocket streams, or tests.
  • EXAMPLES.md - runnable snippets from Hello World to multi-backend wiring.

Installation

For a quick start from PyPI:

pip install lib_log_rich

Detailed installation options (venv, pipx, uv, Poetry/PDM, and Git installs) live in INSTALL.md. If you plan to enable systemd journald logging, follow the extra host prerequisites in INSTALL_JOURNAL.md.

Journald adapter dependency

The Journald adapter requires the systemd-python bindings. Install via the optional extra:

# Requires libsystemd-dev system package to build
sudo apt-get install libsystemd-dev   # Debian/Ubuntu
pip install lib_log_rich[journald]

Alternatively, use your distro's pre-built package (sudo apt-get install python3-systemd on Debian/Ubuntu) which avoids the build step.

Without the bindings, enable_journald=True still works on a systemd host: the adapter falls back to writing the journal's native datagram format to /run/systemd/journal/socket (then /dev/log). The bindings remain the supported path. A write that no journal socket accepts fails that one sink, which the runtime reports as an adapter_error diagnostic while the other sinks still receive the event. The adapter sets no SYSLOG_IDENTIFIER, so journalctl -t <name> finds nothing from it; query by field instead: journalctl SERVICE=<service> -o verbose. See INSTALL_JOURNAL.md for a deeper walkthrough covering Linux service managers and verification steps.

Windows Event Log adapter dependency

The Windows Event Log adapter calls win32evtlogutil.ReportEvent from pywin32. Install it via the optional extra (the marker makes it a no-op off Windows):

pip install lib_log_rich[eventlog]

Without pywin32, init(RuntimeConfig(..., enable_eventlog=True)) still succeeds, but every Event Log write fails with RuntimeError("pywin32 is required for Windows Event Log support"), reported as an adapter_error diagnostic.

JSON encoder and free-threaded Python

Free-threaded (no-GIL) Python builds:

BuildSupportedNotes
3.15tyestested in CI on Linux and Windows
3.14tyestested in CI on Linux and Windows
3.13tnodoes not install: pydantic-core and orjson ship no 3.13t build, and pydantic-core's build tooling (PyO3) dropped 3.13t

Events, checkpoints and JSON dumps are encoded with orjson where it can run, and with the standard library's json otherwise. orjson has no free-threaded build, so on a free-threaded interpreter lib_log_rich never imports it, and the GIL stays off. Below Python 3.14 orjson is installed with the package. From 3.14, the first release with supported free-threading, it is the optional orjson extra, because no installer marker can tell a free-threaded build from a normal one; on a normal 3.14 or later add it for the faster encoder:

pip install "lib_log_rich[orjson]"   # normal (GIL) Python 3.14+; never on a free-threaded build

Both encoders write the same JSON, with orjson's choices for the values the standard library treats differently: null for NaN and infinity, ISO text for datetimes, a UUID's string form, an Enum's value. Two differences remain, valid JSON either way: a float exponent is spelled 1e-07 rather than 1e-7, and an int beyond 64 bits is written out where orjson refuses it.

Why we recommend uv: supply-chain protection

This repository commits a uv.lock. It pins every package the project needs, transitive dependencies included, to an exact version and to the hashes of that version's files. The exclude-newer setting under [tool.uv] in pyproject.toml adds a cooldown: uv leaves any release younger than that out of the lock. CI installs from the lock, so a hijacked release that gets pulled again within the cooldown never reaches our builds, and a file whose hash does not match is refused.

That protection stays in this repository. The installable package declares only lower bounds (>=) for its dependencies, so a plain pip install takes the newest release of every transitive dependency at the moment you run it. If one of them was hijacked an hour earlier, you get the hijacked version. pipx install and uv tool install without options do the same.

To install the command line tool with a cooldown of your own (uv then skips any release younger than 7 days, transitive ones included), or to run the exact tree our CI tests:

uv tool install --exclude-newer "7 days" lib_log_rich
git clone https://github.com/bitranox/lib_log_rich.git && cd lib_log_rich && uv sync --locked && uv run lib_log_rich --help

If you depend on lib_log_rich in your own project, give that project the same cooldown (7 days here, as an example) and commit the uv.lock that uv add lib_log_rich writes:

[tool.uv]
exclude-newer = "7 days"

Logging with LoggerProxy

import lib_log_rich as log

config = log.RuntimeConfig(
    service="my-service",
    environment="dev",
    queue_enabled=False,
    enable_graylog=False,
)
log.init(config)

with log.bind(job_id="startup", request_id="req-001"):
    logger = log.getLogger("app.http")
    logger.info("ready", extra={"port": 8080})

# Inspect the recent history (text/json/html_table/html_txt)
print(log.dump(dump_format="json"))

log.shutdown()

How to use in submodules

Initialise the runtime once near your process entrypoint, then let every module fetch its own LoggerProxy via lib_log_rich.getLogger. The runtime keeps the Rich-backed pipeline global, so submodules only need the name they wish to log under.

# entrypoint.py
import lib_log_rich as log

log.init(log.RuntimeConfig(service="billing", environment="prod"))

from billing import payments  # import after init so the runtime exists

payments.PaymentProcessor().charge_order("ord-42")

log.shutdown()  # drain the queue so every record reaches its sinks

# billing/payments.py
from __future__ import annotations

from collections.abc import Callable
from lib_log_rich import bind, getLogger
from lib_log_rich.runtime import LoggerProxy

LoggerFactory = Callable[[str], LoggerProxy]


class PaymentProcessor:
    def __init__(self, get_logger: LoggerFactory | None = None) -> None:
        self._get_logger = get_logger or getLogger

    def charge_order(self, order_id: str) -> None:
        logger = self._get_logger(__name__)
        with bind(extra={"order_id": order_id}):
            logger.info("Submitting charge", extra={"provider": "stripe"})


# Composition root
processor = PaymentProcessor()
processor.charge_order("ord-123")
  • Key points:
    • No module-level mutable state; PaymentProcessor takes a LoggerFactory, so tests can supply a stub while production code uses lib_log_rich.getLogger.
    • The binding is scoped inside the method, which is explicit and test-friendly.
    • You can register a singleton instance if desired (processor = PaymentProcessor()), yet the module stays stateless and aligns with the repo's dependency-injection guidelines.

If you just want a module-level logger but can not guarantee lib_log_rich.init(...) ran before the module is imported, add a tiny helper that initialises on demand once and then hands back the cached proxy. The helper keeps the module stateless and makes reuse in notebooks or ad-hoc scripts painless while still respecting the single-runtime rule.

# billing/payments.py
from __future__ import annotations

from lib_log_rich import RuntimeConfig, bind, getLogger, init
from lib_log_rich.runtime import LoggerProxy, is_initialised


def ensure_logging() -> LoggerProxy:
    if not is_initialised():
        try:
            init(RuntimeConfig(service="billing", environment="prod"))
        except RuntimeError:
            # Another thread/process section won the race; re-check before propagating.
            if not is_initialised():
                raise
    return getLogger(__name__)


logger: LoggerProxy = ensure_logging()


def charge_order(order_id: str) -> None:
    with bind(extra={"order_id": order_id}):
        logger.info("Submitting charge", extra={"provider": "stripe"})

If you are certain lib_log_rich.init(...) already executed (for example in your CLI entrypoint), you can replace ensure_logging() with a direct getLogger(__name__) assignment and keep the rest identical.

Domain helpers

The domain package (lib_log_rich.domain) exposes reusable value objects so you can keep adapters and feature modules decoupled from implementation modules. A few shortcuts you might want immediately:

import logging
from datetime import datetime, timezone

from lib_log_rich.domain import DumpFormat, LogContext, LogEvent, LogLevel, build_dump_filter

# Translate stdlib levels into the richer domain enum (icons, display metadata, etc.).
domain_level = LogLevel.from_python_level(logging.WARNING)

# Convert back when bridging to the stdlib logging module.
python_level = domain_level.to_python_level()

# Parse dump format values coming from CLI flags or environment variables.
dump_format = DumpFormat.from_name("json")

# Build reusable filters and exercise them against recorded events.
filters = build_dump_filter(context={"service": "billing"}, extra={"tenant": "acme"})
event = LogEvent(
    event_id="evt-1",
    timestamp=datetime.now(tz=timezone.utc),
    logger_name="billing.worker",
    level=LogLevel.INFO,
    message="ready",
    context=LogContext(service="billing", environment="prod", job_id="worker-1", extra={"tenant": "acme"}),
    extra={"tenant": "acme", "order_id": "ORD-7"},
)
should_emit = filters.matches(event)
  • LogLevel keeps conversions idempotent (from_name, from_python_level, to_python_level, and coerce, which accepts a member, a name or a stdlib integer), so threading a standard logging.LogRecord level through Rich adapters only needs a single call.
  • DumpFormat.from_name(...) parses human-friendly inputs ("json", "html_table", etc.) and keeps the call site self-documenting.
  • build_dump_filter(...) returns a DumpFilter you can reuse in unit tests, notebook exploration, or dump pipelines by invoking matches(...) or handing its field tuples to the runtime façade.
  • LoggerProxy.log(level, msg, *args, exc_info=None, stack_info=False, stacklevel=1, extra=None) mirrors the stdlib logging.Logger signature while still normalising enum/string/integer levels. Messages are formatted lazily inside the process pipeline. exc_info can be True, an exception instance or a (type, value, traceback) tuple (ready-made traceback text goes in extra={"exc_info": text}; a string given as the exc_info argument is ignored), and stack_info=True captures the current stack; both are stored as text on the event (see Exceptions logging). The stacklevel keyword is accepted for API parity and currently ignored.
  • LoggerProxy.exception(msg, *args, exc_info=True, stack_info=False, stacklevel=1, extra=None) matches logging.Logger.exception: it logs at LogLevel.ERROR, defaults exc_info to True (capturing the active exception), and otherwise delegates to the same structured pipeline as .error(...).
  • LoggerProxy.setLevel(level) updates the proxy's own threshold. Events below that level are discarded before entering the processing pipeline, so a message must pass the proxy level and each handler's level (console/backend/Graylog) to be emitted.

Custom pipeline wiring

Need to instrument specific collaborators (for example, injecting fakes in a benchmark or swapping adapters without reinitialising the runtime)? Assemble the process pipeline directly with the public dependency bundle:

from datetime import datetime, timedelta, timezone
from rich.console import Console

from lib_log_rich import application
from lib_log_rich.adapters import RegexScrubber, RichConsoleAdapter, SlidingWindowRateLimiter
from lib_log_rich.application import ProcessPipelineDependencies
from lib_log_rich.domain import ContextBinder, LogLevel, RingBuffer, SeverityMonitor
from lib_log_rich.domain.identity import SystemIdentity
from lib_log_rich.runtime import PayloadLimits


class StubClock:
    def now(self) -> datetime:
        return datetime.now(tz=timezone.utc)


class StubIdProvider:
    def __call__(self) -> str:
        return "evt-stub"


class StubIdentity:
    def resolve_identity(self) -> SystemIdentity:
        return SystemIdentity(user_name="demo", hostname="example", process_id=1234)


binder = ContextBinder()
dependencies = ProcessPipelineDependencies(
    context_binder=binder,
    ring_buffer=RingBuffer(max_events=64),
    severity_monitor=SeverityMonitor(),
    console=RichConsoleAdapter(console=Console(record=True), no_color=True),
    console_level=LogLevel.INFO,
    structured_backends=(),
    backend_level=LogLevel.INFO,
    graylog=None,
    graylog_level=LogLevel.ERROR,
    scrubber=RegexScrubber(patterns={}),
    rate_limiter=SlidingWindowRateLimiter(max_events=10, interval=timedelta(seconds=1)),
    clock=StubClock(),
    id_provider=StubIdProvider(),
    limits=PayloadLimits(),
    identity=StubIdentity(),
)

process = application.create_process_log_event(dependencies)
with binder.bind(service="custom", environment="demo", job_id="example"):
    payload = process(
        logger_name="example.pipeline",
        level=LogLevel.INFO,
        message="hello",
        extra={"scope": "demo"},
    )

assert payload.ok is True

# The stub classes above satisfy the relevant ports (`ClockPort`, `IdProvider`,
# `SystemIdentityPort`) so you can compose the pipeline without touching
# runtime internals. In production you would rely on the factories exported via
# ``lib_log_rich.application`` instead.

Context vs. per-event metadata

lib_log_rich.bind(...) establishes the context for subsequent log calls: service, environment, job/request identifiers, trace/span IDs, user, hostname, PID, and optional LogContext.extra. The extra argument on bind is a stable mapping you want attached to every event in that scope (for example, deployment labels or tenant metadata). Every event emitted inside the bound scope inherits the entire context automatically.

The extra= argument on logger.debug/info/... supplements a single event with ad-hoc details (order IDs, feature flags, timing data). The runtime merges the per-event extra with the bound context to form the structured payload that adapters see.

Payload limits apply to both buckets: context extras are capped at 20 keys/256 characters, while per-event extras default to 25 keys/512 characters with depth and aggregate guards. Oversized values are truncated (with a …[truncated] suffix) and the optional diagnostic hook receives events such as extra_keys_dropped, extra_value_truncated_depth_collapsed, or context_extra_keys_dropped when clamping occurs. Nested mappings deeper than extra_max_depth are collapsed into JSON strings so adapters (and downstream storage) are spared from unbounded recursion.

For example:

import lib_log_rich as log

log.init(log.RuntimeConfig(service="billing", environment="prod", queue_enabled=False))

# Context-wide extras travel with every event in the scope
with log.bind(
    service="billing",
    environment="prod",
    job_id="invoice-processor",
    extra={"deployment": "blue", "team": "finops"},
):
    logger = log.getLogger("billing.worker")
    # Per-event extras describe this specific message
    logger.info(
        "processed invoice",
        extra={"invoice_id": "INV-42", "duration_ms": 183},
    )

# The emitted event includes:
#   context.extra -> {"deployment": "blue", "team": "finops"}
#   event.extra   -> {"invoice_id": "INV-42", "duration_ms": 183}

log.shutdown()

Exceptions logging

.exception(...), or exc_info=True, an exception instance or a (type, value, traceback) tuple on any level helper, records the formatted traceback text that traceback.format_exception produces, as stdlib logging does. Stdlib records arriving through attach_std_logging() are recorded the same way. Ready-made traceback text passed as extra["exc_info"] moves out of extra into the event's exc_info field without being re-formatted (a string given as the exc_info argument itself is silently ignored), and stack_info=True records the formatted stack:

import json

import lib_log_rich as log

log.init(log.RuntimeConfig(service="svc", environment="dev", queue_enabled=False))
logger = log.getLogger("billing.worker")

try:
    raise RuntimeError("upstream failed")
except RuntimeError:
    logger.exception("job crashed")  # records the formatted traceback

logger.warning("slow call", stack_info=True)  # records the formatted stack

events = json.loads(log.dump(dump_format=log.DumpFormat.JSON))
print(events[0]["exc_info"])
log.shutdown()

The text is stored on LogEvent.exc_info / LogEvent.stack_info, and every sink shows it, exc_info first, as stdlib logging does: the Rich console and the text and HTML text dumps print it indented under the event line, the HTML table puts it in the message cell, Graylog sends it as GELF full_message, journald appends it to MESSAGE and also sends EXCEPTION_TEXT / STACK_TEXT, and the Windows Event Log adds exc_info=... / stack_info=... strings. JSON dumps and the ring-buffer checkpoint carry both fields. The scrubber does not touch this text, just as it leaves the message alone, so an exception message reaches every sink as written.

Long traces are compacted in two steps, and the diagnostic hook receives exc_info_truncated / stack_info_truncated for each:

  1. stacktrace_max_frames (default 10) counts frames: a File "...", line N, in f line plus the indented source and caret lines under it. When the text holds more than twice that many frames, the first and last stacktrace_max_frames frames are kept and the frames between are removed: each run of removed frames becomes one ... truncated K frame(s) ... line at that run's indentation, and the diagnostic carries frames_removed, the sum of K over all runs. Only frames are removed: every Traceback (most recent call last): header, exception line and separator of a chained exception or an exception group is kept, and so is a [Previous line repeated N more times] line whose frame stays. Text with no frame lines is not frame-compacted.
  2. Text still longer than extra_max_value_chars (default 512) is cut by keeping its end, which for most tracebacks is the innermost frames and the exception line; an exception message longer than the limit fills the kept text on its own. The kept text starts with [truncated]…, and the diagnostic carries reason, original_length, new_length and key. Raise extra_max_value_chars in payload_limits to keep longer traces.

Opt-in .env loading

lib_log_rich honours real environment variables over RuntimeConfig fields (LOG_SERVICE, LOG_CONSOLE_LEVEL, and friends). The .env helpers let you keep that precedence while sourcing defaults from a project-local file:

import lib_log_rich as log
import lib_log_rich.config as log_config

log_config.enable_dotenv()  # walk upwards from cwd, load the first .env found
config = log.RuntimeConfig(service="svc", environment="dev", queue_enabled=False)
log.init(config)
...
log.shutdown()

Key points:

  • .env loading is explicit - nothing is read unless you call enable_dotenv().
  • Precedence stays intact: CLI flag -> real os.environ -> discovered .env -> defaults.
  • Search walks up from the cwd (or search_from) and stops at the first .env, at a directory holding a pyproject.toml/.git marker, or at a directory it cannot look into. A .env that is not UTF-8 raises DotenvDecodeError, naming the file and byte offset but none of its content.
  • Pass dotenv_override=True when you intentionally want .env values to win over real environment variables.

See DOTENV.md for more detail, examples, and CLI usage.

Checking a configuration without starting a runtime

validate_config(config) raises the ValueError that init(config) would raise, with the same message, but never starts a runtime and never touches one that is already running. Use it to judge a candidate configuration (a reloaded config file, a --profile switch, an edited .env) while logging keeps working, instead of shutting down and initialising again to find out.

import lib_log_rich as log

candidate = log.RuntimeConfig(service="svc", environment="dev", console_level="loud")
try:
    log.validate_config(candidate)
except ValueError as exc:
    print(exc)  # Invalid runtime settings: console_level: Unknown log level: 'loud'

Key points:

  • init() and validate_config() resolve settings through the same code, so they cannot disagree: a config one accepts, the other accepts.
  • Environment overrides (LOG_*, including a loaded .env) apply exactly as they would for init(), so the verdict holds for an init() made in the same environment.
  • Refused before any adapter is built: unknown level names (in the config or in LOG_CONSOLE_LEVEL / LOG_BACKEND_LEVEL / LOG_GRAYLOG_LEVEL), scrub patterns that do not compile (config or LOG_SCRUB_PATTERNS), an unknown name in LOG_CONSOLE_THEME, LOG_CONSOLE_STREAM, LOG_CONSOLE_FORMAT_PRESET, LOG_DUMP_FORMAT_PRESET or LOG_GRAYLOG_PROTOCOL, the console-only short_loc_icon as dump preset, a ring_buffer_size that is not positive (or a LOG_RING_BUFFER_SIZE that is not an integer), a malformed LOG_GRAYLOG_ENDPOINT or LOG_RATE_LIMIT, Graylog enabled without an endpoint, UDP combined with TLS while Graylog is enabled, a ring_buffer_checkpoint that is a directory, an empty service or environment (including an empty LOG_SERVICE / LOG_ENVIRONMENT), and, while the built-in console is used, a console style key that is not a level name. A custom console_adapter_factory decides for itself what style keys mean, so those are not checked then.
  • An unparsable LOG_QUEUE_MAXSIZE, LOG_QUEUE_FULL_POLICY, LOG_QUEUE_PUT_TIMEOUT or LOG_QUEUE_STOP_TIMEOUT is not refused: it is ignored and the configured value stays.
  • An unknown theme, Graylog protocol, queue policy, console stream or format preset, and an invalid payload_limits value, never reach validate_config(): RuntimeConfig(...) itself refuses them with a pydantic ValidationError (a ValueError subclass) naming the field.
  • RuntimeConfig(...) also refuses a keyword that is not one of its fields, so a misspelled setting (ring_bufer_size=5) raises a ValidationError naming it instead of leaving the intended value at its default.
  • Two refusals are outside its reach, because it builds no adapters and reads no files: an adapter from console_adapter_factory that lacks a ConsolePort method (emit or flush), which init() refuses with a TypeError naming the missing method, and a ring_buffer_checkpoint file holding a line it cannot load, which init() refuses with a ValueError naming the file and line.
  • A refusal reads Invalid runtime settings: <reason>, with the field or LOG_ variable in front of the reason when a single one is to blame (console_level: Unknown log level: 'loud'); several problems in the settings models are joined with ; .

Integration Options

Stdlib compatibility & deliberate differences

LoggerProxy implements the core logging.Logger surface so existing call sites rarely need adjustment. Supported helpers are:

  • .debug(...), .info(...), .warning(...), .error(...), .critical(...)
  • .exception(...) (defaults exc_info=True just like the stdlib)
  • .log(level, ...)
  • .setLevel(level)

Key differences from the stdlib logging framework:

  • Calls return a ProcessResult describing the emission outcome instead of None.
  • No hierarchical logger tree: we do not expose the root logger, getChild(), propagation, or handler management. Each getLogger(name) call yields an independent proxy whose name is metadata only.
  • Aliases such as .warn(...), .fatal(...), isEnabledFor(...), addHandler(...), and handler lists are intentionally unsupported.
  • stacklevel is accepted but currently ignored. Stack trace capture is explicit: provide exc_info or stack_info to record traces. The runtime stores the formatted, compacted text on the event, and every sink shows it (see exceptions logging).
  • Console output defaults to stderr (matching the standard library logger), but you can steer it via RuntimeConfig.console_stream to ConsoleStream.STDOUT, ConsoleStream.STDERR, ConsoleStream.BOTH, ConsoleStream.CUSTOM (supply console_stream_target), or ConsoleStream.NONE to suppress console output entirely (RuntimeConfig also parses the value strings "stdout", "stderr", "both", "custom" and "none"; ConsoleStream comes from lib_log_rich.domain). You can also pass a no-op console_adapter_factory for bespoke mute behaviour. Without a console sink nothing touches stdio; events continue to flow through the configured backends.

When migrating code that relies on root/sub-logger behaviour or handler mutation, refactor to request proxies explicitly via lib_log_rich.getLogger(name) and configure sinks through RuntimeConfig instead of the stdlib globals.

Integrating stdlib logging

Existing projects often expose module-level loggers via logging.getLogger(). The runtime ships with a bridge so those callers can continue to emit without refactoring.

import logging

from lib_log_rich.runtime import RuntimeConfig, attach_std_logging, init

init(RuntimeConfig(service="legacy-app", environment="prod", queue_enabled=False, enable_graylog=False))
attach_std_logging()  # logger_level defaults to get_minimum_log_level()

logging.getLogger(__name__).info("legacy path", extra={"tenant": 42})

if __name__ == "__main__":
    from lib_log_rich.runtime import dump

    print(dump())
  • attach_std_logging(...) registers a single StdlibLoggingHandler on the chosen logger (root by default). By default it sets the logger level to get_minimum_log_level() so events aren't pre-filtered before reaching lib_log_rich, and sets propagate=False to prevent duplicate emission. Pass logger_level=None to leave the level unchanged.
  • Every logging.LogRecord is translated into the runtime pipeline: message/arguments flow through the same sanitiser as LoggerProxy, extra metadata travels unchanged, and a record's exc_info/stack_info is stored as formatted text exactly as for LoggerProxy (see Exceptions logging).
  • Location metadata (pathname, lineno, funcName, plus any custom extra fields) is copied into the event so dumps, Graylog, and Rich console output can display the original call site.
  • Records originating from lib_log_rich are kept out of the pipeline to avoid recursion. When the bridge is the only handler such a record would reach, warnings and errors (an adapter that failed, for example) go to stderr through logging.lastResort, exactly as stdlib prints them when no handler is configured; with a handler of your own on the chain, that handler prints them instead. Set extra={"lib_log_rich_skip": True} if you need to suppress specific third-party records; those are dropped silently.

Understanding get_minimum_log_level()

attach_std_logging() automatically sets the logger level to get_minimum_log_level(), which returns the lowest (most permissive) threshold among active adapters. This ensures stdlib doesn't pre-filter events before they reach lib_log_rich.

import logging
import lib_log_rich as log
from lib_log_rich.runtime import attach_std_logging

# Initialize with independent levels for different adapters
config = log.RuntimeConfig(
    service="my-service",
    environment="prod",
    console_level="INFO",  # Console shows INFO and above
    backend_level="WARNING",  # Journald/EventLog shows WARNING and above
    graylog_level="ERROR",  # Graylog shows ERROR and above
    enable_graylog=True,
    graylog_endpoint=("graylog.example.com", 12201),
)
log.init(config)
attach_std_logging()  # Automatically uses get_minimum_log_level() -> INFO

# Now stdlib loggers won't pre-filter
logging.getLogger("app.module").info("This reaches lib_log_rich console")
logging.getLogger("app.module").debug("This is filtered by lib_log_rich, not stdlib")

log.shutdown()

Key points:

  • Independent levels: console_level, backend_level, and graylog_level are completely independent. Each gates events to its respective adapter without affecting the others.
  • get_minimum_log_level(): Returns the lowest (most permissive) threshold among active adapters. backend_level counts only when journald or the Windows Event Log is enabled, and graylog_level only when Graylog is enabled.
  • Automatic integration: attach_std_logging() uses this by default; pass logger_level=None to leave the stdlib logger level unchanged.

Runtime Reference

Public API

All runtime configuration flows through RuntimeConfig. Create an instance with the desired fields and pass it to lib_log_rich.init(config). The table below summarises the fields and related helpers.

SymbolSignature (abridged)Description
initinit(config: RuntimeConfig)Composition root. Wires adapters, queue, scrubber, and rate limiter. Pass a RuntimeConfig instance; environment variables listed below continue to override matching fields.
validate_configvalidate_config(config: RuntimeConfig) -> NoneRaises the ValueError init() would raise for config, with the same message, without starting a runtime or touching a running one. Environment overrides apply exactly as for init(). Use it to check a candidate configuration while logging is running instead of shutting down and initialising again.
bindbind(*, service=None, environment=None, job_id=None, request_id=None, user_id=None, user_name=None, hostname=None, process_id=None, process_id_chain=None, trace_id=None, span_id=None, extra=None) (context manager)Binds contextual metadata for the scope. The fields are keyword-only; anything else goes in extra (for example bind(extra={"order_id": order_id})), and an unknown field raises TypeError. None inherits the parent frame's value; init() primes a base frame from the config's service and environment with job_id="bootstrap", so a scope names only what it adds or overrides. Yields the active LogContext.
getLoggergetLogger(name: str) -> LoggerProxyReturns a LoggerProxy exposing .debug/.info/.warning/.error/.critical/.exception. Each call returns a ProcessResult (e.g. ProcessResult(ok=True, event_id="...") or ProcessResult(ok=False, reason=DropReason.RATE_LIMITED)).
LoggerProxycreated via getLogger(name)Lightweight facade around the process use case. Level helpers mirror the stdlib signature: .debug(msg, *args, exc_info=None, stack_info=False, stacklevel=1, extra=None) (and similarly for info/warning/error/critical) plus .exception(msg, *args, exc_info=True, ...), .log(level, msg, *args, ...), and .setLevel(level). Messages are formatted in the pipeline, exc_info/stack_info are stored on the event as formatted traceback/stack text (see Exceptions logging), and stacklevel is accepted but ignored. .setLevel(...) updates the proxy's own threshold so events must satisfy it and the handler levels; calls still return a ProcessResult (unlike logging.Logger, which returns None).
dumpdump(*, dump_format: str | DumpFormat = "text", path=None, level: str | LogLevel | None = None, console_format_preset: FormatPreset | None = None, console_format_template=None, theme: ConsoleTheme | None = None, console_styles=None, context_filters=None, context_extra_filters=None, extra_filters=None, color=False) -> strSerialises the ring buffer (DumpFormat.TEXT, JSON, HTML_TABLE, HTML_TXT, or their value strings). theme and console_format_preset take enum members only; a plain string raises TypeError. level filters events by severity, presets/templates customise text rendering (template wins; FormatPreset.SHORT_LOC_ICON is console-only and a text dump refuses it with ValueError), theme/console_styles reuse or override the runtime palette, the context_*/extra_* filter mappings narrow results by metadata, and color toggles ANSI output for text dumps. Payloads are always returned and optionally written to path.
get_minimum_log_levelget_minimum_log_level() -> LogLevelReturns the minimum (most permissive) log level among all active adapters: the lowest of console_level, backend_level (only when journald or the Windows Event Log is enabled) and graylog_level (only when Graylog is enabled). Useful for setting the stdlib logging.Logger root level to match the lowest threshold, ensuring no events are pre-filtered before reaching lib_log_rich. Raises RuntimeError if called before init().
max_level_seenmax_level_seen() -> LogLevel | NoneReturns the highest severity observed since the runtime was initialised or metrics were reset, or None before any event. Handy for "only dump logs when something >= ERROR happened" checks.
severity_snapshotseverity_snapshot() -> SeveritySnapshotCaptures totals, per-level counts, threshold buckets, and drop statistics (by reason and severity). Safe to call from any thread.
reset_severity_metricsreset_severity_metrics() -> NoneClears severity and drop counters without touching the ring buffer or adapters. Invoke after you've handed dumps to operators.
shutdownshutdown() -> NoneFlushes adapters, drains and stops the queue, persists the ring buffer to its checkpoint (when ring_buffer_checkpoint is set), and clears global state. Calling it again before another init() raises RuntimeError, as does calling it inside a running event loop (await shutdown_async() there).
shutdown_asyncasync shutdown_async() -> NoneAsync variant of shutdown() for code running in an event loop (shutdown() raises RuntimeError there). It waits for the queue worker on a worker thread, so a diagnostic hook that hands work back to the loop (call_from_thread, run_coroutine_threadsafe(...).result()) cannot deadlock it.
flushflush(timeout: float | None = None, *, flush_ring_buffer: bool = False) -> NoneDrains queues and flushes all adapters (console, Graylog) without terminating the runtime. Unlike shutdown(), logging remains active after this call. Raises TimeoutError if the queue doesn't drain within timeout (default: the configured queue_stop_timeout, or 5.0 s when that is None). flush_ring_buffer=True appends the buffer's events to its checkpoint and clears the buffer, when the runtime has one: init() configures it from ring_buffer_checkpoint (env LOG_RING_BUFFER_CHECKPOINT); without it flush_ring_buffer=True does nothing and the buffer is preserved. Raises RuntimeError if called from within an active event loop.
flush_asyncflush_async(timeout: float | None = None, *, flush_ring_buffer: bool = False) -> NoneAsync variant of flush(). Awaitable from async contexts. Same behaviour: drains queue, flushes adapters, keeps runtime active. Raises TimeoutError on queue drain timeout. It waits for the queue worker on a worker thread, so a diagnostic hook that hands work back to the loop (call_from_thread, run_coroutine_threadsafe(...).result()) cannot deadlock it.
hello_worldhello_world() -> NonePrints the canonical "Hello World" message for smoke tests.
i_should_faili_should_fail() -> NoneRaises RuntimeError("I should fail") to exercise failure handling paths.
summary_infosummary_info() -> strReturns the CLI metadata banner as a string without printing it.
logdemologdemo(*, theme=ConsoleTheme.CLASSIC, service=None, environment=None, dump_format: DumpFormat | None = None, dump_path=None, color=None, console_format_preset: FormatPreset | None = None, console_format_template=None, dump_format_preset: FormatPreset | None = None, dump_format_template=None, context_filters=None, context_extra_filters=None, extra_filters=None, enable_graylog=False, graylog_endpoint=None, graylog_protocol=GraylogProtocol.TCP, graylog_tls=False, enable_journald=False, enable_eventlog=False) -> LogDemoResultSpins up a temporary runtime, emits one sample event per level, optionally renders a dump, and records which backends were requested in LogDemoResult.backends (a BackendStatus with graylog, journald and eventlog flags). The result also carries theme, styles, events (one ProcessResult per level), dump, service and environment. Use the boolean flags to exercise Graylog, journald, or Windows Event Log sinks from the CLI or API.
SeveritySnapshotfrozen dataclass returned by severity_snapshot()Fields: highest, total_events, counts, thresholds, dropped_total, drops_by_reason (keyed by DropReason), drops_by_level, drops_by_reason_and_level (keyed by (DropReason, LogLevel)). All mappings are read-only copies so you can serialise them directly.
QueueConsoleAdapterQueueConsoleAdapter(queue, *, export_style=ExportStyle.ANSI, force_color=False, no_color=False, styles=None, format_preset: FormatPreset | None = None, format_template=None, console_width=None)Threaded console adapter (import from lib_log_rich.runtime or lib_log_rich.adapters.console) that renders Rich output into a queue.Queue. It reuses the Rich formatter so console level thresholds and styling stay consistent. export_style and format_preset take enum members; a plain string raises TypeError. Pass via console_adapter_factory to stream ANSI or HTML lines to GUIs, SSE/WebSocket feeds, or background workers without touching global state.
AsyncQueueConsoleAdapterAsyncQueueConsoleAdapter(queue, *, export_style=ExportStyle.ANSI, force_color=False, no_color=False, styles=None, format_preset: FormatPreset | None = None, format_template=None, console_width=None, on_drop=None, loop=None)Asyncio variant (import from lib_log_rich.runtime or lib_log_rich.adapters.console) targeting asyncio.Queue producers/consumers. It shares the Rich formatter and level gate with the default console adapter. Ideal for Textual apps, async servers, or tests that await console output; wire it with console_adapter_factory alongside the threaded adapter. Chunks are enqueued with put_nowait, so full queues drop latest segments; on_drop receives each dropped segment (without it the adapter logs a warning). An emit from another thread (the runtime queue worker) is handed to the queue's loop with call_soon_threadsafe; that loop is the one running at construction, or loop=.
ExportStyleExportStyle.ANSI, ExportStyle.HTMLString enum (from lib_log_rich.domain or lib_log_rich.runtime) selecting the payload format returned by queue-backed console adapters (ANSI for terminal passthrough, HTML for rich web panes).
attach_std_loggingattach_std_logging(*, logger=None, handler_level=None, logger_level=<minimum>, propagate=False) -> StdlibLoggingHandlerImport from lib_log_rich.runtime. Registers one StdlibLoggingHandler on logger (the root logger by default) and returns it. handler_level sets the handler's own level; logger_level defaults to get_minimum_log_level() and None leaves the logger level unchanged; propagate sets the logger's propagate flag. See Integrating stdlib logging.
ConsoleTheme, DumpFormat, FormatPreset, GraylogProtocolenumsImportable from lib_log_rich as well as from lib_log_rich.domain. dump() and logdemo() take theme and the preset parameters as enum members only.

LoggerProxy methods and logdemo

LoggerProxy instances returned by getLogger() support the standard logging-level methods, and every call that passes the proxy's own level feeds the runtime's severity monitor (see Inspecting severity and drop metrics).

import lib_log_rich as log

config = log.RuntimeConfig(service="svc", environment="metrics", queue_enabled=False)
log.init(config)

with log.bind(job_id="run-42"):
    log.getLogger("svc.worker").info("started")
    log.getLogger("svc.worker").error("boom")

# `LoggerProxy` instances returned by `getLogger()` support the standard logging-level methods:

logger = log.getLogger("app.component")
logger.info("payload %s", "alice", extra={"user": "alice"})
logger.error("boom", exc_info=True)
logger.warning("slow call", stack_info=True)
try:
    raise RuntimeError("bad input")
except RuntimeError:
    logger.exception("captured runtime failure")

# Drop INFO/DEBUG at the proxy before they hit any handlers
logger.setLevel("ERROR")

Each call returns a ProcessResult describing the outcome (ok plus event_id, queued=True, or ok=False with a reason such as DropReason.RATE_LIMITED). Standard logging APIs return None; this diagnostic payload is the deliberate divergence we keep for caller observability.

The public .log(...) helper normalises level inputs from strings ("warning"), integers (logging.WARNING), or the domain enum so advanced callers can apply dynamic thresholds without reimplementing conversions. Format strings are interpolated inside the process pipeline, so %-style placeholders behave exactly like logging.Logger. When exc_info or stack_info are provided they are stored on the event as formatted text, compacted according to the configured payload limits (see Exceptions logging). The stacklevel keyword is accepted for drop-in compatibility but currently ignored. .exception(...) is a thin convenience wrapper over .error(...) that sets level=LogLevel.ERROR and defaults exc_info to True. .setLevel(...) gates events at the proxy, leaving console/backend/Graylog thresholds untouched so both conditions must be satisfied for emission.

The optional extra mapping is copied into the structured event and travels end-to-end: it is scrubbed, persisted in the ring buffer, and forwarded to every adapter (Rich console, journald, Windows Event Log, Graylog, dump exporters). Use it to attach contextual fields such as port numbers, tenant IDs, or feature flags. journald names each field after its key, uppercased and mapped to the characters journald accepts: A-Z, 0-9 and _, at most 64 long, not starting with a digit or _ (order-id becomes ORDER_ID, _x becomes EXTRA__X); a name that clashes with a reserved or already-set field gets the EXTRA_ prefix.

Need a quick preview of console colours? Call:

import lib_log_rich as log

result = log.logdemo(theme=log.ConsoleTheme.NEON, dump_format=log.DumpFormat.JSON)
print(result.events)  # one ProcessResult per level
print(result.dump)  # rendered dump string (or None when not requested)
print(result.backends)  # BackendStatus(graylog=False, journald=False, eventlog=False)

The helper initialises a throwaway runtime, emits one message per level using the selected theme, optionally renders a text/JSON/HTML dump via the dump_format argument, and then shuts itself down. Themes are the ConsoleTheme members CLASSIC, DARK, NEON, and PASTEL, described in CONSOLESTYLES.md. ConsoleTheme, DumpFormat, FormatPreset and GraylogProtocol are importable from lib_log_rich itself as well as from lib_log_rich.domain. logdemo() and dump() take theme and the preset parameters as enum members only; a plain string such as "neon" raises TypeError. console_styles overrides individual level styles on top of a theme.

The optional backend flags (enable_graylog, enable_journald, enable_eventlog) let you route the demo events to real adapters during manual testing - the result exposes the chosen targets via result.backends.

Runtime configuration

RuntimeConfig holds every runtime setting (keyword arguments only; an unknown keyword is refused); pass it to lib_log_rich.init(config). Environment variables in the last column override the matching field.

ParameterTypeDefaultExpected valuesPurposeEnvironment variable
servicestr(required)Non-empty identifier such as checkout, worker, billing.Logical service name recorded in each event and used by adapters.LOG_SERVICE
environmentstr(required)Deployment label (dev, stage, prod, local, ...)Deployment environment recorded in each event and used by adapters.LOG_ENVIRONMENT
console_levelstr | LogLevelLogLevel.INFOCase-insensitive debug, info, warning, error, critical or a LogLevel enum.Lowest level emitted to the Rich console adapter. Independent of backend/Graylog levels.LOG_CONSOLE_LEVEL
backend_levelstr | LogLevelLogLevel.WARNINGSame set as console_level.Threshold shared by journald and Windows Event Log adapters. Independent of console/Graylog levels.LOG_BACKEND_LEVEL
graylog_endpointtuple[str, int] | NoneNone(host, port) tuple (port > 0); only LOG_GRAYLOG_ENDPOINT takes a HOST:PORT string.Host/port for GELF; combine with enable_graylog=True.LOG_GRAYLOG_ENDPOINT (host:port)
graylog_protocolGraylogProtocolGraylogProtocol.TCPGraylogProtocol.TCP or GraylogProtocol.UDP; the value strings tcp/udp parse in any case.Transport to reach Graylog.LOG_GRAYLOG_PROTOCOL
graylog_tlsboolFalseTrue only with GraylogProtocol.TCP.Wrap the Graylog connection in TLS.LOG_GRAYLOG_TLS
graylog_levelstr | LogLevelLogLevel.WARNINGSame set as console_level.Severity threshold for Graylog fan-out. Independent of console/backend levels. Use get_minimum_log_level() to find the lowest among all three.LOG_GRAYLOG_LEVEL
enable_ring_bufferboolTrueTrue/False.Toggles the in-memory ring buffer (falls back to 1024 events when disabled).LOG_RING_BUFFER_ENABLED
ring_buffer_sizeint25_000Integer > 0 (events).Max events retained in the ring buffer.LOG_RING_BUFFER_SIZE
ring_buffer_checkpointPath | NoneNoneA file path; an empty string means none; an existing directory is refused at init().JSONL file the ring buffer is loaded from at init() and appended to by flush(flush_ring_buffer=True), shutdown() and the flush after a dump. It only grows: rotate it yourself. A corrupt line makes init() raise ValueError.LOG_RING_BUFFER_CHECKPOINT
enable_journaldboolFalseTrue/False; effective on Linux/systemd only.Adds the journald adapter.LOG_ENABLE_JOURNALD
enable_eventlogboolFalseTrue/False; effective on Windows only.Adds the Windows Event Log adapter.LOG_ENABLE_EVENTLOG
enable_graylogboolFalseTrue/False; True without an endpoint is refused.Enables the Graylog adapter.LOG_ENABLE_GRAYLOG
queue_enabledboolTrueTrue/False.Routes events through the background queue worker.LOG_QUEUE_ENABLED
queue_maxsizeint2048Integer > 0 (slots).Maximum pending events before the full-policy applies.LOG_QUEUE_MAXSIZE
queue_full_policyQueuePolicyQueuePolicy.BLOCKQueuePolicy.BLOCK or QueuePolicy.DROP; the value strings block/drop parse.Choose whether producers wait or drop when the queue is full.LOG_QUEUE_FULL_POLICY
queue_put_timeoutfloat | None1.0Seconds > 0, or None/<=0 for indefinite wait.Timeout applied to blocking queue puts before falling back to the caller.LOG_QUEUE_PUT_TIMEOUT
queue_stop_timeoutfloat | None5.0Seconds > 0, or None/<=0 to wait forever.Deadline for draining the queue during shutdown().LOG_QUEUE_STOP_TIMEOUT
force_colorboolFalseTrue/False.Forces Rich colour output even when stderr is not a TTY.LOG_FORCE_COLOR
no_colorboolFalseTrue/False.Disables colour output regardless of terminal support.LOG_NO_COLOR
console_stylesMapping[str | LogLevel, str] | NoneNoneMapping keyed by DEBUG, INFO, WARNING, ERROR, CRITICAL (or LogLevel members) with Rich style strings.Rich style overrides per level.LOG_CONSOLE_STYLES (comma-separated LEVEL=style)
console_themeConsoleTheme | NoneConsoleTheme.DARKConsoleTheme.CLASSIC, DARK, NEON or PASTEL; value strings parse in any case, a blank string is None, and an unknown name is refused when RuntimeConfig is built.Palette applied to the console (and inherited by dumps without overrides).LOG_CONSOLE_THEME
console_format_presetFormatPreset | Noneplatform-specificFormatPreset.FULL, SHORT, FULL_LOC, SHORT_LOC, SHORT_LOC_ICON; value strings parse in any case. None means SHORT_LOC_ICON on Windows, SHORT_LOC on Linux/Mac.Console line layout when no custom template is supplied.LOG_CONSOLE_FORMAT_PRESET
console_format_templatestr | NoneNonePython str.format template using fields listed in Template fields.Custom console template overriding the preset.LOG_CONSOLE_FORMAT_TEMPLATE
console_streamConsoleStreamConsoleStream.STDERRConsoleStream.STDOUT, STDERR, BOTH, CUSTOM or NONE (value strings parse): stdout, stderr, both, a custom stream, or mute entirely.Default mirrors stdlib logging (stderr); tweak to match host expectations.LOG_CONSOLE_STREAM
console_stream_targetobject | NoneNoneFile-like object implementing write(); required with ConsoleStream.CUSTOM.Inject in-memory buffers or other sinks without a custom adapter factory.(none)
console_adapter_factoryCallable[[ConsoleAppearance], ConsolePort] | NoneNoneCallable returning a custom ConsolePort adapter.Plug-in point for queue-backed or test adapters; see STREAMINGCONSOLE.md.(none)
dump_format_presetFormatPreset | NoneNone (FormatPreset.FULL)FormatPreset.FULL, SHORT, FULL_LOC, SHORT_LOC; the console-only SHORT_LOC_ICON is refused at init().Default text dump layout used when callers omit one.LOG_DUMP_FORMAT_PRESET
dump_format_templatestr | NoneNonePython str.format template using Template fields.Default text dump template overriding the preset.LOG_DUMP_FORMAT_TEMPLATE
scrub_patternsdict[str, str] | None{"password": ".+", "secret": ".+", "token": ".+"}Mapping of field=regex pairs merged with built-in scrub rules.Removes sensitive values before fan-out.LOG_SCRUB_PATTERNS
rate_limittuple[int, float] | NoneNone(max_events, window_seconds) with both positives; env string MAX:WINDOW_SECONDS.Throttles emissions before fan-out.LOG_RATE_LIMIT
payload_limitsPayloadLimits | NoneSee belowA PayloadLimits, or a mapping of its fields (message_max_chars, extra_max_keys, etc.) validated into one when RuntimeConfig is built, so an invalid limit is refused there.Guards message/extra size; see Payload Limits.(none)
diagnostic_hookCallable[[str, Mapping[str, object]], None] | NoneNoneCallable receiving (event_name, diagnostics) or None.Observes queue drops, rate limiting, payload clamps, etc.(none)
config.enable_dotenv() helper(call before init())(opt-in)Optional search_from, markers, dotenv_override; see DOTENV.md.Loads .env files while preserving env precedence.LOG_USE_DOTENV (CLI/module entry points)

Console template fields

console_format_template and dump_format_template accept the same Python str.format placeholders. Common fields include:

PlaceholderDescription
timestampISO 8601 UTC timestamp with microseconds.
timestamp_trimmed_naiveUTC timestamp trimmed to whole seconds without timezone information.
timestamp_locHost-local ISO 8601 timestamp with offset.
LEVELUpper-case level name (DEBUG, INFO, WARNING, ERROR, CRITICAL).
level_iconRich glyph representing the level (bug, info, warning, cross, skull).
level_codeFour-character abbreviation (DEBG, INFO, WARN, ERRO, CRIT).
logger_nameName passed to getLogger(...) when the event was emitted.
messageLog message string.
context_fieldsSpace-prefixed key=value pairs merged from LogContext and extra.
user_name, hostname, process_idSystem identity fields captured during initialisation.
process_id_chainParent/child process lineage formatted as pid1>pid2>pid3.
extraShallow copy of the extra mapping supplied to the logger.

See LOGDUMP.md for the exhaustive placeholder list (including individual date/time components and dotted aliases).

See STREAMINGCONSOLE.md for advanced queue-backed console wiring examples that build on console_adapter_factory.

Payload Limits

These guards exist to keep a single buggy or malicious caller from flooding the ring buffer, queue, or downstream sinks with giant payloads. The defaults clamp events to dimensions that safely fit journald, GELF, and log-shipper expectations while still leaving room for rich context.

Pass a PayloadLimits instance (from lib_log_rich.runtime) as payload_limits, for example:

import lib_log_rich as log
from lib_log_rich.runtime import PayloadLimits

config = log.RuntimeConfig(
    service="svc",
    environment="prod",
    payload_limits=PayloadLimits(message_max_chars=2048, extra_max_keys=10),
)
log.init(config)

RuntimeConfig also accepts a mapping of the same fields (from a config file, say) and validates it into a PayloadLimits when it is built, so an invalid limit, or a key that is not one of these fields, is refused there with a ValidationError.

Default limits guard the pipeline and can be tuned per environment. Every field is optional; unspecified values fall back to the defaults below.

PayloadLimits fields

  • truncate_message (bool, default True) - when True long messages are truncated to message_max_chars; when False oversized messages raise ValueError.
  • message_max_chars (int, default 4096) - maximum characters for the primary log message.
  • extra_max_keys (int, default 25) - maximum number of keys accepted in the extra mapping attached to the event. Additional keys are dropped with a diagnostic hook notice.
  • extra_max_value_chars (int, default 512) - per-key character cap after values are stringified; excess content is truncated with a …[truncated] suffix. exc_info and stack_info text is capped at the same length but keeps its end instead, starting with [truncated]… (see Exceptions logging).
  • extra_max_depth (int, default 3) - nesting depth allowed before nested structures are stringified.
  • extra_max_total_bytes (int | None, default 8192) - total UTF-8 encoded size allowed for the sanitized extra payload. Set to None to disable the aggregate clamp.
  • context_max_keys (int, default 20) - maximum keys stored in LogContext.extra.
  • context_max_value_chars (int, default 256) - per-value limit for context metadata once stringified.
  • stacktrace_max_frames (int, default 10) - number of leading and trailing frames preserved in exc_info and stack_info text once it holds more than twice that many. A frame is a File "...", line N, in f line plus the indented lines under it; the frames between are removed, each run of them replaced with one ... truncated N frame(s) ... line, and the result is then subject to extra_max_value_chars.

Whenever a limit is enforced, the optional diagnostic_hook receives an event (for example message_truncated, extra_keys_dropped, exc_info_truncated) so operators can monitor clamping in production. Queue resilience signals use the same channel: queue_worker_error fires when the background worker raises (and the adapter flips its worker_failed flag for health checks, clearing automatically after the cooldown or a clean restart), and queue_drop_callback_error captures drop-handler failures without tearing the worker down.

Graylog fan-out uses the configured graylog_level (default WARNING when enabled, automatically tightened to CRITICAL when Graylog is disabled). Text dumps do not inherit the console preset or template: they use dump_format_preset / dump_format_template (default FormatPreset.FULL) unless a dump() call passes its own.

The initializer also honours LOG_BACKEND_LEVEL, LOG_FORCE_COLOR, and LOG_NO_COLOR simultaneously: a set, valid environment variable wins over the matching RuntimeConfig field. When enable_journald is requested on Windows hosts or enable_eventlog on non-Windows hosts the runtime silently disables those adapters so cross-platform deployments never fail during initialisation.

Note: TLS is only supported with the TCP transport. With Graylog enabled, combining GraylogProtocol.UDP with TLS (or setting LOG_GRAYLOG_PROTOCOL=udp alongside LOG_GRAYLOG_TLS=1) raises a ValueError during initialisation.

Environment-only overrides

Set these, restart your process, and the runtime will merge them with the RuntimeConfig you pass to init(...).

VariableDefaultExpected valuesEffect
LOG_SERVICERuntimeConfig.serviceNon-empty string (e.g., checkout).Override the advertised service name.
LOG_ENVIRONMENTRuntimeConfig.environmentNon-empty string (e.g., dev, stage, prod).Override the deployment/stage label.
LOG_CONSOLE_LEVELinfodebug, info, warning, error, critical (case-insensitive).Minimum level emitted to the console adapter.
LOG_BACKEND_LEVELwarningSame as LOG_CONSOLE_LEVEL.Threshold for journald/Event Log adapters.
LOG_GRAYLOG_LEVELwarningSame as LOG_CONSOLE_LEVEL.Threshold for Graylog emission.
LOG_RING_BUFFER_ENABLEDtrueBoolean toggle (1/true/yes/on or 0/false/no/off).Disable to skip ring-buffer retention.
LOG_RING_BUFFER_SIZE25000Integer > 0 (events).Resize the in-memory ring buffer.
LOG_RING_BUFFER_CHECKPOINTunsetA file path; empty means unset.Persist the ring buffer to this JSONL file (beats ring_buffer_checkpoint).
LOG_ENABLE_JOURNALDfalseBoolean toggle (ignored on Windows).Enable/disable the journald adapter.
LOG_ENABLE_EVENTLOGfalseBoolean toggle (ignored on non-Windows).Enable/disable the Windows Event Log adapter.
LOG_ENABLE_GRAYLOGfalseBoolean toggle; requires LOG_GRAYLOG_ENDPOINT.Enable the Graylog adapter.
LOG_GRAYLOG_ENDPOINTnoneHOST:PORT string with port > 0.Target host and port for GELF.
LOG_GRAYLOG_PROTOCOLtcptcp or udp; anything else is refused at init().Choose Graylog transport.
LOG_GRAYLOG_TLSfalseBoolean toggle; only valid with tcp.Wrap the Graylog connection in TLS.
LOG_QUEUE_ENABLEDtrueBoolean toggle.Disable to process fan-out inline without a queue.
LOG_QUEUE_MAXSIZE2048Integer > 0 (slots).Queue capacity before the full-policy applies.
LOG_QUEUE_FULL_POLICYblockblock or drop; an unknown value falls back to the configured policy.Decide whether producers wait or drop when the queue is full.
LOG_QUEUE_PUT_TIMEOUT1.0Seconds (float). <=0 clears the timeout.Timeout for blocking puts before the caller handles overflow.
LOG_QUEUE_STOP_TIMEOUT5.0Seconds (float). <=0 waits indefinitely.Drain deadline during shutdown.
LOG_FORCE_COLORfalseBoolean toggle.Force ANSI colour even when stderr is not a TTY.
LOG_NO_COLORfalseBoolean toggle.Strip colour output entirely.
LOG_CONSOLE_THEMEdarkclassic, dark, neon, pastel (case-insensitive); an unknown name is refused at init(), an empty value keeps the configured theme.Apply a palette to the console/dumps.
LOG_CONSOLE_STYLESnoneComma-separated LEVEL=style pairs (e.g., INFO=green).Override Rich styles per level.
LOG_CONSOLE_FORMAT_PRESETplatform-specificfull, short, full_loc, short_loc, short_loc_icon; an unknown name is refused at init(). Default: short_loc_icon on Windows, short_loc on Linux/Mac.Default preset for console lines.
LOG_CONSOLE_FORMAT_TEMPLATEnonestr.format template using Template fields.Override the preset with a custom layout.
LOG_CONSOLE_STREAMstderrstdout, stderr, both, custom, none; anything else is refused at init().Redirect console output; custom requires console_stream_target, none mutes the console.
LOG_DUMP_FORMAT_PRESETfullfull, short, full_loc, short_loc; an unknown name, or the console-only short_loc_icon, is refused at init().Default preset when dumping with dump_format="text".
LOG_DUMP_FORMAT_TEMPLATEnonestr.format template using Template fields.Custom text-dump template.
LOG_SCRUB_PATTERNSpassword=.+,secret=.+,token=.+Comma-separated field=regex pairs.Merge additional scrub patterns with defaults.
LOG_RATE_LIMITnoneMAX:WINDOW_SECONDS with positive numbers (e.g., 500:60).Rate limit emissions before fan-out.
LOG_USE_DOTENVfalseBoolean toggle.Allow the CLI/module entry point to load a nearby .env.

lib_log_rich stresstest also pre-fills its form from LOG_DUMP_FORMAT and LOG_DUMP_LEVEL; the runtime itself ignores them.

Queue safety defaults

See also QUEUE.md for a complete guide to queue policies and diagnostics.

A blocking put waits up to one second by default (queue_put_timeout=1.0); None, or LOG_QUEUE_PUT_TIMEOUT set to <=0, waits indefinitely. When the timeout is positive and the worker remains in a failed state, blocking puts fall back to drop mode once the timeout elapses and the runtime emits a queue_degraded_drop_mode diagnostic. After recovery (cooldown expiry, clean stop(drain=True), or a fresh start()), the adapter restores the configured policy. Tune queue_put_timeout or LOG_QUEUE_PUT_TIMEOUT if you need bounded waits.

Boolean variables accept 1, true, yes, on as true and 0, false, no, off as false (case-insensitive); an empty or unrecognised value keeps the configured value.

Operations & Observability

Log dump

log.dump(...) bridges the in-memory ring buffer to structured exports. See LOGDUMP.md for parameter tables, text placeholder references, and usage notes covering text/JSON/HTML dumps.

JSON dumps expose enriched metadata (level_name, numeric level_value, the four-character level_code, and the console level_icon) plus a normalised process_id_chain. When you need to isolate specific events, provide mapping-based filters such as context_filters={"job_id": "batch-42"} or extra_filters={"request": {"icontains": "api"}}. Entries accept exact values, substring predicates (contains/icontains), or regex dictionaries ({"pattern": r"^prefix", "regex": True}), and multiple keys combine with logical AND while repeated keys OR together.

Inspecting severity and drop metrics

The runtime keeps a thread-safe severity monitor so you can make dump decisions without scanning the ring buffer.

import lib_log_rich as log
from lib_log_rich.domain import DropReason, LogLevel

log.init(log.RuntimeConfig(service="svc", environment="metrics", queue_enabled=False))
with log.bind(job_id="run-42"):
    log.getLogger("svc.worker").info("started")
    log.getLogger("svc.worker").error("boom")

snapshot = log.severity_snapshot()
assert snapshot.highest is LogLevel.ERROR
assert snapshot.total_events == 2
assert snapshot.dropped_total == 0
assert snapshot.drops_by_reason[DropReason.RATE_LIMITED] == 0  # counters are pre-seeded
warning_or_higher = snapshot.thresholds[LogLevel.WARNING]
assert warning_or_higher == 1

# Drops caused by rate limiting, queue saturation, or adapter failures
# surface by reason and severity:

rate_limited_errors = snapshot.drops_by_reason.get(DropReason.RATE_LIMITED, 0)

log.reset_severity_metrics()  # start a fresh window without reinitialising the runtime

if log.max_level_seen() is None:
    print("No high-severity events since the reset")

log.shutdown()

drops_by_reason is pre-seeded with the DropReason members RATE_LIMITED, QUEUE_FULL and ADAPTER_ERROR (from lib_log_rich.domain), and a QUEUE_FAILURE drop (the queue refused the event) adds that key. ProcessResult.reason can also be LOGGER_LEVEL (below the proxy's setLevel threshold); those events never enter the pipeline and are not counted. Use drops_by_reason_and_level when you need per-severity breakdowns for dashboards or alerts.

Threshold buckets complement the per-level counts: each threshold represents an "at least" view (for example the built-in LogLevel.WARNING bucket includes WARNING, ERROR, and CRITICAL events). The runtime seeds WARNING and ERROR by default so operators can monitor actionable activity without summing individual levels. When wiring a monitor manually (e.g., for custom tooling) you can pass additional LogLevel values to SeverityMonitor(thresholds=...) to track different cut-offs.

  • By default the SeverityMonitor tracks two cumulative thresholds:

    • LogLevel.WARNING: counts every event logged at WARNING, ERROR, or CRITICAL.
    • LogLevel.ERROR: counts every event logged at ERROR or CRITICAL.

    Those buckets are included automatically so you can see "actionable" volume without summing individual levels. If you need different cut-offs (e.g., include INFO or add a CRITICAL-only bucket) you can pass your own iterable when constructing the monitor - SeverityMonitor(thresholds=[LogLevel.INFO, LogLevel.ERROR]) - and the snapshot will expose exactly those you request.

Terminal compatibility

Rich automatically detects whether the target is 16-colour, 256-colour, or truecolor, and adjusts the style to the nearest supported palette. For truly minimal environments (plain logs, CI artefacts), set no_color=True (or LOG_NO_COLOR=1) and Rich suppresses ANSI escapes entirely. Conversely, force_color=True (or LOG_FORCE_COLOR=1) forces colouring even if stderr isn't a tty (useful in some container setups).

Console & Terminal Experience

Customising per-level colours

Override the default Rich styles by passing a dictionary as RuntimeConfig(console_styles=...) or by exporting LOG_CONSOLE_STYLES as a comma-separated list, for example:

export LOG_CONSOLE_STYLES="DEBUG=dim,INFO=bright_green,WARNING=bold yellow,ERROR=bold white on red,CRITICAL=bold magenta"

Values use Rich's style grammar (named colours, modifiers like bold/dim, or hex RGB). Omitted keys fall back to the built-in theme. logdemo cycles through all preset x theme combinations (5 presets x 4 themes = 20 examples by default) so you can preview styles before committing to overrides. Use --preset and --theme options to filter specific combinations.

CLI and Automation

CLI entry point

lib_log_rich ships with a rich-click interface for quick diagnostics, demos, and automation. See CLI.md for the full command breakdown, option tables, and usage examples. Quick highlight: run python -m lib_log_rich for the metadata banner, use lib_log_rich logdemo to preview console themes and generate text/JSON/HTML dumps (with optional Graylog, journald, or Event Log fan-out), or launch lib_log_rich stresstest to drive a Textual TUI that stress-tests the runtime while streaming live diagnostics. (The stress tester requires Textual; install dev extras with pip install -e .[dev] before running it.) Filtering options such as --context-exact job_id=batch and --extra-regex request=^api flow through logdemo so CLI dumps can focus on specific workloads without post-processing. Regex-based filters validate patterns eagerly and raise a friendly click.BadParameter when a pattern is invalid (exit code 2), so a typo never surfaces as a raw re.error traceback.

Streaming console output to other consumers

RuntimeConfig.console_adapter_factory lets you swap the Rich console for a custom adapter without touching internals. Ship a queue-backed adapter (see lib_log_rich.adapters.console.QueueConsoleAdapter) to feed GUIs, SSE/WebSocket endpoints, or tests. Each adapter honours the same format presets/templates and level filtering as the Rich console and can emit ANSI strings or HTML snippets. A detailed walk-through with threaded, asyncio, and composite patterns lives in STREAMINGCONSOLE.md.

A minimal Flask example lives in examples/flask_console_stream.py; it pushes rendered HTML lines to an async EventSource stream while regular CLI/TUI usage keeps the Textual stress tester responsive - no monkey patching required.

Quick smoke-test helpers ship with the package:

import lib_log_rich as log

log.hello_world()
try:
    log.i_should_fail()
except RuntimeError as exc:
    print(exc)

Claude Code skill

This repo is itself a Claude Code plugin marketplace and ships a skill (skills/python-logging/) that teaches a coding agent to install, configure, deploy and operate lib_log_rich: RuntimeConfig and its LOG_* environment overrides, the console, journald, Windows Event Log and Graylog sinks, systemd, container and Windows service deployment, secret scrubbing, the stdlib bridge, queues and subprocesses, validate_config, ring-buffer dumps and the CLI. Install it in any project:

/plugin marketplace add bitranox/lib_log_rich
/plugin install lib_log_rich

The same skill is also available in the central bitranox-skills marketplace as coding-python-logging.

Further documentation

Development

Contributor workflows, make targets, CI automation, and release guidance are documented in DEVELOPMENT.md.

License

MIT

↑↓ navigate openesc close