Structured Logging API Reference¶

class DropEvent(Exception):
    """Raise inside a processor to silently discard the current log event."""

class PrintLogger:
    """Minimal logger that writes to a file handle via ``print()``.

    This is the default underlying logger.  It has no filtering, no
    handler chain -- just ``print()`` to the configured stream.

    Args:
        file: Output stream.  Defaults to ``sys.stderr``.
    """

    def __init__(self, file: IO[str] | None = None) -> None:
        self._file = file or sys.stderr
        self._write = self._file.write
        self._flush = self._file.flush

    def msg(self, message: str) -> None:
        """Write *message* followed by a newline."""
        self._write(message + "\n")
        self._flush()

    debug = info = warning = warn = error = critical = fatal = msg

class PrintLoggerFactory:
    """Factory that creates :class:`PrintLogger` instances.

    Args:
        file: Output stream passed to each ``PrintLogger``.
    """

    def __init__(self, file: IO[str] | None = None) -> None:
        self._file = file

    def __call__(self, *args: Any) -> PrintLogger:
        return PrintLogger(file=self._file)

class StdlibLoggerFactory:
    """Factory that returns a :class:`logging.Logger` from the stdlib.

    Args:
        name: Logger name passed to ``logging.getLogger()``.
            If *None*, uses the root logger.
    """

    def __init__(self, name: str | None = None) -> None:
        self._name = name

    def __call__(self, *args: Any) -> logging.Logger:
        if args:
            return logging.getLogger(args[0])
        return logging.getLogger(self._name)

class TimeStamper:
    """Processor that adds a timestamp to the event dict.

    Args:
        fmt: Timestamp format.  ``"iso"`` for ISO 8601, ``None`` for a
            UNIX float, or a :func:`~datetime.datetime.strftime` format
            string.  Defaults to ``"iso"``.
        utc: If *True*, use UTC; otherwise local time.
        key: Dict key for the timestamp.  Defaults to ``"timestamp"``.
    """

    __slots__ = ("_fmt", "_utc", "_key")

    def __init__(
        self,
        fmt: str | None = "iso",
        utc: bool = True,
        key: str = "timestamp",
    ) -> None:
        self._fmt = fmt
        self._utc = utc
        self._key = key

    def __call__(
        self, logger: Any, method_name: str, event_dict: EventDict
    ) -> EventDict:
        now = (
            datetime.datetime.now(datetime.timezone.utc)
            if self._utc
            else datetime.datetime.now()
        )
        if self._fmt is None:
            event_dict[self._key] = now.timestamp()
        elif self._fmt == "iso":
            event_dict[self._key] = now.isoformat()
        else:
            event_dict[self._key] = now.strftime(self._fmt)
        return event_dict

class KeyValueRenderer:
    """Render the event dict as ``key=value`` pairs.

    Args:
        key_order: Keys to render first, in this order.
        sort_keys: Sort remaining keys alphabetically.
        drop_missing: Skip *key_order* keys that are absent from the dict.
    """

    __slots__ = ("_key_order", "_sort_keys", "_drop_missing")

    def __init__(
        self,
        key_order: list[str] | None = None,
        sort_keys: bool = False,
        drop_missing: bool = True,
    ) -> None:
        self._key_order = key_order or []
        self._sort_keys = sort_keys
        self._drop_missing = drop_missing

    def __call__(self, logger: Any, method_name: str, event_dict: EventDict) -> str:
        ordered: list[tuple[str, Any]] = []
        remaining = dict(event_dict)

        for key in self._key_order:
            if key in remaining:
                ordered.append((key, remaining.pop(key)))
            elif not self._drop_missing:
                ordered.append((key, None))

        rest = sorted(remaining.items()) if self._sort_keys else list(remaining.items())
        ordered.extend(rest)

        parts = [
            f"{k}={v!r}" if not isinstance(v, str) else f"{k}={v}" for k, v in ordered
        ]
        return " ".join(parts)

class JSONRenderer:
    """Render the event dict as a JSON string.

    Args:
        serializer: JSON serialization function.  Defaults to
            :func:`json.dumps`.
        **dumps_kw: Extra keyword arguments passed to *serializer*.
    """

    __slots__ = ("_serializer", "_dumps_kw")

    def __init__(
        self,
        serializer: Callable[..., str] = _json.dumps,
        **dumps_kw: Any,
    ) -> None:
        self._serializer = serializer
        self._dumps_kw = dumps_kw
        if "default" not in self._dumps_kw:
            self._dumps_kw["default"] = _json_default

    def __call__(self, logger: Any, method_name: str, event_dict: EventDict) -> str:
        return self._serializer(event_dict, **self._dumps_kw)

class ConsoleRenderer:
    """Render the event dict as colorized, loguru-style console output.

    Output format::

        2026-03-27 14:30:00.123 | INFO     | event message    key=val key=val

    Args:
        colors: Enable ANSI color codes.  *None* auto-detects terminal
            support.
        pad_event: Pad the event field to this width for alignment.
        level_styles: Override per-level ANSI color strings.
    """

    __slots__ = ("_colors", "_pad_event", "_level_styles")

    def __init__(
        self,
        colors: bool | None = None,
        pad_event: int = 30,
        level_styles: dict[str, str] | None = None,
    ) -> None:
        self._colors = _supports_color() if colors is None else colors
        self._pad_event = pad_event
        self._level_styles = level_styles or _LEVEL_STYLES

    def __call__(self, logger: Any, method_name: str, event_dict: EventDict) -> str:
        # Extract well-known keys
        event = str(event_dict.pop("event", ""))
        level = event_dict.pop("level", method_name)
        timestamp = event_dict.pop("timestamp", None)
        event_dict.pop("logger", None)
        exception = event_dict.pop("exception", None)

        # Build timestamp string
        if timestamp is None:
            now = datetime.datetime.now()
            ms = now.microsecond // 1000
            ts_str = now.strftime("%Y-%m-%d %H:%M:%S.") + f"{ms:03d}"
        elif isinstance(timestamp, str):
            # Try to format ISO timestamps more compactly
            ts_str = _compact_iso(timestamp)
        elif isinstance(timestamp, float):
            dt = datetime.datetime.fromtimestamp(timestamp)
            ts_str = dt.strftime("%Y-%m-%d %H:%M:%S.") + f"{dt.microsecond // 1000:03d}"
        else:
            ts_str = str(timestamp)

        # Build level label
        level_str = _LEVEL_LABELS.get(level, f"{level.upper():<8s}")

        # Build key=value pairs from remaining context
        kv_parts: list[str] = []
        for k, v in event_dict.items():
            kv_parts.append(f"{k}={v!r}" if not isinstance(v, str) else f"{k}={v}")
        kv_str = " ".join(kv_parts)

        # Pad event
        padded_event = event.ljust(self._pad_event) if kv_str else event

        # Assemble
        if self._colors:
            msg_color = self._level_styles.get(level, _Colors.WHITE)
            label_color = _LEVEL_LABEL_STYLES.get(level, _Colors.WHITE)

            line = (
                f"{_Colors.GREEN}{ts_str}{_Colors.RESET} | "
                f"{label_color}{_Colors.BOLD}{level_str}{_Colors.RESET} | "
                f"{msg_color}{padded_event}{_Colors.RESET}"
            )
            if kv_str:
                line += f" {_Colors.DIM}{kv_str}{_Colors.RESET}"
            if exception:
                line += f"\n{_Colors.RED}{exception}{_Colors.RESET}"
        else:
            line = f"{ts_str} | {level_str} | {padded_event}"
            if kv_str:
                line += f" {kv_str}"
            if exception:
                line += f"\n{exception}"

        return line

class BoundLogger:
    """A logger that carries bound context through a processor pipeline.

    Do not instantiate directly; use :func:`get_logger` or
    :func:`wrap_logger` instead.

    Args:
        logger: The underlying logger instance (e.g. ``PrintLogger``).
        processors: Ordered list of processors to run on each log event.
        context: Initial context dictionary.
    """

    __slots__ = ("_logger", "_processors", "_context")

    def __init__(
        self,
        logger: Any,
        processors: list[Processor],
        context: dict[str, Any],
    ) -> None:
        self._logger = logger
        self._processors = processors
        self._context = context

    def bind(self, **new_values: Any) -> BoundLogger:
        """Return a new logger with *new_values* merged into the context."""
        new_ctx = {**self._context, **new_values}
        return BoundLogger(self._logger, self._processors, new_ctx)

    def unbind(self, *keys: str) -> BoundLogger:
        """Return a new logger with *keys* removed from the context."""
        new_ctx = {k: v for k, v in self._context.items() if k not in keys}
        return BoundLogger(self._logger, self._processors, new_ctx)

    def new(self, **new_values: Any) -> BoundLogger:
        """Return a new logger with *new_values* replacing the context."""
        return BoundLogger(self._logger, self._processors, dict(new_values))

    # ── Log methods ──

    def debug(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("debug", event, kw)

    def info(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("info", event, kw)

    def warning(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("warning", event, kw)

    def warn(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("warning", event, kw)

    def error(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("error", event, kw)

    def critical(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("critical", event, kw)

    def fatal(self, event: str | None = None, /, **kw: Any) -> None:
        self._process("critical", event, kw)

    def exception(self, event: str | None = None, /, **kw: Any) -> None:
        kw.setdefault("exc_info", True)
        self._process("error", event, kw)

    def log(self, level: int, event: str | None = None, /, **kw: Any) -> None:
        method_name = _LEVEL_TO_NAME.get(level, "info")
        self._process(method_name, event, kw)

    # ── Internal ──

    def _process(self, method_name: str, event: str | None, kw: dict[str, Any]) -> None:
        """Merge context, run processors, and emit via the underlying logger."""
        event_dict: EventDict = {**self._context, **kw}
        if event is not None:
            event_dict["event"] = event
        elif "event" not in event_dict:
            event_dict["event"] = ""

        try:
            for proc in self._processors:
                event_dict_or_str = proc(self._logger, method_name, event_dict)
                if isinstance(event_dict_or_str, str):
                    # Final renderer returned a string -- emit and stop.
                    _emit(self._logger, method_name, event_dict_or_str)
                    return
                event_dict = event_dict_or_str
        except DropEvent:
            return

        # If no processor returned a string, emit the repr of event_dict.
        _emit(self._logger, method_name, str(event_dict))