gitversioned.logging¶

def autolog(
    func: _FuncT | None = None,
    *,
    exception_log_level: str | None = "ERROR",
) -> _FuncT | Callable[[_FuncT], _FuncT]:
    """
    Decorate a function to automatically log call inputs, outputs, and
    raised exceptions.

    This decorator logs inputs before execution, logs the return value on
    success, and logs any raised exceptions with trace details before
    re-raising.

    Example:
        .. code-block:: python

            from gitversioned.logging import autolog

            @autolog
            def calculate_sum(a: int, b: int) -> int:
                return a + b

    :param func: Target function to wrap, defaults to None.
    :param exception_log_level: Log level for exception reporting, defaults to "ERROR".
    :returns: The decorated wrapper or a decorator factory function.
    """

    def decorator(func_to_wrap: _FuncT) -> _FuncT:
        @functools.wraps(func_to_wrap)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            func_name = getattr(func_to_wrap, "__qualname__", "function")
            logger.debug(
                _LOG_ENTRY_FORMAT.format(name=func_name, args=args, kwargs=kwargs)
            )
            try:
                result = func_to_wrap(*args, **kwargs)
            except Exception as error:
                if exception_log_level == "ERROR":
                    logger.opt(exception=error).error(
                        _LOG_EXCEPTION_FORMAT.format(name=func_name, exception=error),
                    )
                elif exception_log_level is not None:
                    logger.log(
                        exception_log_level,
                        _LOG_EXCEPTION_FORMAT.format(name=func_name, exception=error),
                    )
                raise error
            else:
                logger.debug(_LOG_EXIT_FORMAT.format(name=func_name, result=result))
                return result

        return cast("_FuncT", wrapper)

    if func is None:
        return decorator
    return decorator(func)

def configure_logger(  # noqa: C901, PLR0912, PLR0915
    settings: LoggingSettings | None = None,
    **default_overrides: Any,
) -> None:
    """
    Configure the global Loguru logger based on settings.

    This function initializes or updates the active logger handler. If no
    settings are provided, it loads settings from environment variables.
    It enables interception of standard library log statements and configures
    formatting, sinks, filtering, and queue options.

    Example:
        .. code-block:: python

            from gitversioned.logging import LoggingSettings, configure_logger

            configure_logger(
                settings=LoggingSettings(level="INFO"),
                clear_loggers=True
            )

    :param settings: Logging configurations, defaults to None (loads from environment).
    :param default_overrides: Parameter overrides merged into env settings if
                              settings is None.
    :returns: None.
    :raises ImportError: Raised if OpenTelemetry formatting is enabled but the
                         package is not installed.
    """
    if settings is None:
        env_settings = LoggingSettings()
        merged = default_overrides.copy()
        for field in env_settings.model_fields_set:
            merged[field] = getattr(env_settings, field)
        settings = LoggingSettings(**merged)

    if not settings.enabled:
        logger.disable("gitversioned")
        intercept_standard_logging(False)
        if isinstance(_state["handler_id"], int):
            with contextlib.suppress(ValueError):
                logger.remove(_state["handler_id"])
            _state["handler_id"] = None
        return

    logger.enable("gitversioned")
    intercept_standard_logging(True)

    if settings.clear_loggers:
        if hasattr(logger, "_mock_name") or type(logger).__name__ in (
            "MagicMock",
            "Mock",
        ):
            logger.remove()
        else:
            for handler_id, handler in list(cast("Any", logger)._core.handlers.items()):  # noqa: SLF001
                sink = getattr(handler, "_sink", None)
                handler_obj = getattr(sink, "_handler", None)
                if handler_obj and type(handler_obj).__name__ == "PropagateHandler":
                    continue
                with contextlib.suppress(ValueError):
                    logger.remove(handler_id)
        _state["handler_id"] = None
    elif isinstance(_state["handler_id"], int):
        with contextlib.suppress(ValueError):
            logger.remove(_state["handler_id"])
        _state["handler_id"] = None

    use_otel = settings.otel_formatting == "enable" or (
        settings.otel_formatting == "auto" and opentelemetry_trace is not None
    )
    if settings.otel_formatting == "enable" and opentelemetry_trace is None:
        raise ImportError(
            "OpenTelemetry is not installed but 'otel_formatting' was set to 'enable'."
        )

    if use_otel:
        sink_val = OtelSink(settings.sink)
        log_format = "{message}\n"
    else:
        sink_val = settings.sink
        log_format = settings.format

    filter_val = "gitversioned" if settings.filter is True else settings.filter

    if isinstance(filter_val, str):
        final_filter: Any = filter_val
    elif isinstance(filter_val, (list, tuple)):
        prefixes = tuple(filter_val)

        def final_filter(record: dict[str, Any]) -> bool:
            return bool(record["name"] and record["name"].startswith(prefixes))

    else:
        final_filter = None if filter_val is False else filter_val

    # Resolve "auto" log level
    level_val = settings.level
    if level_val == "auto":
        level_val = "WARNING"

    _state["handler_id"] = logger.add(
        sink=cast("Any", sink_val),
        level=level_val,
        filter=cast("Any", final_filter),
        format=cast("Any", log_format),
        enqueue=settings.enqueue,
        **settings.kwargs,
    )

def intercept_standard_logging(enable: bool = True) -> None:
    """
    Hook standard library logging into or detach it from the Loguru pipeline.

    This function attaches `InterceptHandler` to the standard root logger
    to redirect standard library logs, or removes it to restore the
    previous logging handlers.

    Example:
        .. code-block:: python

            from gitversioned.logging import intercept_standard_logging

            intercept_standard_logging(enable=True)

    :param enable: True to intercept standard logging; False to detach and restore.
    :returns: None.
    """
    root = logging.getLogger()
    if enable:
        if not _standard_handlers:
            handler = InterceptHandler()
            root.addHandler(handler)
            _standard_handlers.append(handler)
    else:
        # Restore a clean environment slate using our tracked handlers
        for handler in _standard_handlers:
            root.removeHandler(handler)
        _standard_handlers.clear()