Source code for ooai_llm.logging

"""Logging helpers for ``ooai_llm``.

Purpose:
    Provide one small logging facade that prefers ``ultilog`` when installed
    and falls back to standard-library logging when the optional dependency is
    absent.
"""

from __future__ import annotations

from collections.abc import Iterator
from contextlib import contextmanager
import importlib
import logging as stdlib_logging
from typing import Any, Literal



LoggingPreset = Literal["dev", "test", "prod"]



LoggingMode = Literal["rich", "plain", "json"]



def _load_ultilog() -> Any | None:
    """Return the optional ``ultilog`` module when available."""
    try:
        return importlib.import_module("ultilog")
    except ImportError:
        return None




def configure_logging(
    *,
    preset: LoggingPreset | None = None,
    level: str | int | None = None,
    mode: LoggingMode | None = None,
    force: bool = False,
    service_name: str | None = None,
    **kwargs: Any,
) -> Any:
    """Configure application logging.

    When ``ultilog`` is installed, this delegates to ``ultilog.setup(...)``.
    Otherwise it configures standard-library logging with ``basicConfig``.
    """
    ultilog = _load_ultilog()
    if ultilog is not None:
        setup = getattr(ultilog, "setup", None)
        if callable(setup):
            options: dict[str, Any] = {"force": force}
            if preset is not None:
                options["preset"] = preset
            if level is not None:
                options["level"] = level
            if mode is not None:
                options["mode"] = mode
            if service_name is not None:
                options["service_name"] = service_name
            options.update(kwargs)
            return setup(**options)

    stdlib_logging.basicConfig(level=level or "INFO", force=force)
    return None





def get_logger(name: str | None = None) -> stdlib_logging.Logger:
    """Return a logger from ``ultilog`` when available, else stdlib logging."""
    ultilog = _load_ultilog()
    if ultilog is not None:
        get_ultilog_logger = getattr(ultilog, "get_logger", None)
        if callable(get_ultilog_logger):
            return get_ultilog_logger(name) if name is not None else get_ultilog_logger()
    return stdlib_logging.getLogger(name or "ooai_llm")



@contextmanager


def logging_context(**fields: Any) -> Iterator[None]:
    """Bind contextual logging fields when ``ultilog`` supports it."""
    ultilog = _load_ultilog()
    if ultilog is not None:
        bind_context = getattr(ultilog, "logging_context", None)
        if callable(bind_context):
            with bind_context(**fields):
                yield
            return
    yield





def log_event(
    logger: stdlib_logging.Logger,
    event: str,
    *,
    level: str = "debug",
    **fields: Any,
) -> None:
    """Emit a structured-ish log event without assuming a specific backend."""
    method = getattr(logger, level, logger.debug)
    extra = {"ooai_event": event}
    extra.update({f"ooai_{key}": value for key, value in fields.items()})
    try:
        method(event, extra=extra)
    except Exception:  # pragma: no cover - logging should never break callers
        method("%s %s", event, fields)