alerts

log_panel.alerts

ThresholdConfig(level, threshold) dataclass

Configuration for a log level threshold.

maybe_emit_threshold_signal(*, sender, logger_name, record_level, timestamp, message, module, pathname, line_number, count_matching_records)

Emit the threshold signal when a log level count crosses its configured limit.

Source code in log_panel/alerts.py

def maybe_emit_threshold_signal(
    *,
    sender: object,
    logger_name: str,
    record_level: str,
    timestamp: datetime,
    message: str,
    module: str,
    pathname: str,
    line_number: int,
    count_matching_records: ThresholdCountCallback,
) -> None:
    """Emit the threshold signal when a log level count crosses its configured limit."""
    threshold_config = get_threshold_config(record_level)
    if threshold_config is None:
        return
    if getattr(_dispatch_local, "dispatching", False):
        return  # pragma: no cover
    if not log_threshold_reached.has_listeners(sender):
        return

    window_end: datetime = timestamp
    window_start: datetime = timestamp - timedelta(hours=1)
    matching_count: int = count_matching_records(
        (threshold_config.level.value,),
        window_start,
        window_end,
    )
    if matching_count != threshold_config.threshold:
        return

    event = ThresholdAlertEvent(
        logger_name=logger_name,
        threshold_level=threshold_config.level,
        record_level=LogLevel(record_level),
        threshold=threshold_config.threshold,
        matching_count=matching_count,
        timestamp=timestamp,
        window_start=window_start,
        window_end=window_end,
        message=message,
        module=module,
        pathname=pathname,
        line_number=line_number,
    )

    try:
        _dispatch_local.dispatching = True
        log_threshold_reached.send_robust(sender=sender, event=event)
    finally:
        _dispatch_local.dispatching = False

get_threshold_config(record_level)

Return the threshold config for record_level, or None if the level is not configured.

Source code in log_panel/alerts.py

def get_threshold_config(record_level: str) -> ThresholdConfig | None:
    """Return the threshold config for *record_level*, or None if the level is not configured."""
    thresholds: dict[str, int | None] = conf.get_thresholds()
    threshold: int | None = thresholds.get(record_level)
    if threshold is None:
        return None
    try:
        level = LogLevel(value=record_level)
    except ValueError:  # pragma: no cover
        return None  # pragma: no cover
    return ThresholdConfig(level=level, threshold=threshold)