managers

log_panel.managers

LogReader

Read-only interface for querying logs outside the admin panel.

Use in your own views, APIs, or background tasks. Subclass and override :meth:get_queryset to apply default filters (e.g. logger name or minimum level restrictions) for a specific user role. Further filters can still be chained on the returned LogQueryset.

get_queryset()

Returns a LogQueryset for the active backend with no filters applied.

Source code in log_panel/managers.py

def get_queryset(self) -> LogQuery:
    """Returns a LogQueryset for the active backend with no filters applied."""
    from log_panel import conf

    return LogQuery(backend=conf.get_backend())

LogRecordManager

Bases: Manager

Manager for the Log model — handles record creation and bulk inserts.

count_threshold_matches(*, logger_name, levels, window_start, window_end)

Count how many log records match the given logger name and level filters within the specified time window.

Source code in log_panel/managers.py

def count_threshold_matches(
    self,
    *,
    logger_name: str,
    levels: tuple[str, ...],
    window_start: datetime,
    window_end: datetime,
) -> int:
    """Count how many log records match the given logger name and level filters within the specified time window."""
    return self.get_queryset().count_threshold_matches(
        logger_name=logger_name,
        levels=levels,
        window_start=to_database_datetime(value=window_start),
        window_end=to_database_datetime(value=window_end),
    )

create_from_record(timestamp, level, logger_name, message, module, pathname, line_number)

Persist a single log record.

Source code in log_panel/managers.py

def create_from_record(
    self,
    timestamp: datetime,
    level: str,
    logger_name: str,
    message: str,
    module: str,
    pathname: str,
    line_number: int,
) -> Log:
    """Persist a single log record."""
    message_parts = self._split_message(message=message)
    log: Log = self.create(
        timestamp=to_database_datetime(value=timestamp),
        level=level,
        logger_name=logger_name,
        message=message_parts.preview,
        message_size=message_parts.size,
        message_chunked=message_parts.is_chunked,
        module=module,
        pathname=pathname,
        line_number=line_number,
    )
    if message_parts.is_chunked:
        from log_panel.models import LogMessageChunk

        LogMessageChunk.objects.db_manager(self.db).bulk_create(
            (
                LogMessageChunk(log=log, index=index, text=chunk)
                for index, chunk in enumerate(message_parts.chunks)
            ),
            batch_size=100,
        )

    db_timestamp: datetime = to_database_datetime(value=timestamp)
    from log_panel.models import LogCard, LogTimelineBucket

    LogCard.objects.db_manager(self.db).upsert(
        logger_name=logger_name,
        total_delta=1,
        error_delta=1 if level in ERROR_LEVELS else 0,
        warning_delta=1 if level == LogLevel.WARNING else 0,
        last_seen=db_timestamp,
    )
    LogTimelineBucket.objects.db_manager(self.db).upsert(
        logger_name=logger_name,
        timestamp=db_timestamp,
        level=level,
    )
    return log

bulk_create_from_records(records)

Persist multiple log records in a single bulk insert operation.

Source code in log_panel/managers.py

def bulk_create_from_records(self, records: list[dict[str, Any]]) -> list[Log]:
    """Persist multiple log records in a single bulk insert operation."""
    from log_panel.models import LogCard, LogMessageChunk, LogTimelineBucket

    parts_list = [self._split_message(message=r["message"]) for r in records]

    log_instances = [
        self.model(
            timestamp=to_database_datetime(value=r["timestamp"]),
            level=r["level"],
            logger_name=r["logger_name"],
            message=parts.preview,
            message_size=parts.size,
            message_chunked=parts.is_chunked,
            module=r["module"],
            pathname=r["pathname"],
            line_number=r["line_number"],
        )
        for r, parts in zip(records, parts_list, strict=True)
    ]

    created_logs: list[Log] = self.bulk_create(log_instances)

    chunk_instances = [
        LogMessageChunk(log=log, index=index, text=chunk)
        for log, parts in zip(created_logs, parts_list, strict=True)
        if parts.is_chunked
        for index, chunk in enumerate(parts.chunks)
    ]
    if chunk_instances:
        LogMessageChunk.objects.db_manager(self.db).bulk_create(
            chunk_instances, batch_size=100
        )

    total_by_logger: Counter[str] = Counter()
    errors_by_logger: Counter[str] = Counter()
    warnings_by_logger: Counter[str] = Counter()
    last_seen_by_logger: dict[str, datetime] = {}
    for r in records:
        name: Any = r["logger_name"]
        total_by_logger[name] += 1
        if r["level"] in ERROR_LEVELS:
            errors_by_logger[name] += 1
        if r["level"] == LogLevel.WARNING:
            warnings_by_logger[name] += 1
        ts: datetime = to_database_datetime(value=r["timestamp"])
        if name not in last_seen_by_logger or ts > last_seen_by_logger[name]:
            last_seen_by_logger[name] = ts

    for name in total_by_logger:
        LogCard.objects.db_manager(self.db).upsert(
            logger_name=name,
            total_delta=total_by_logger[name],
            error_delta=errors_by_logger[name],
            warning_delta=warnings_by_logger[name],
            last_seen=last_seen_by_logger[name],
        )

    LogTimelineBucket.objects.db_manager(self.db).bulk_upsert(records)

    return created_logs

LogCardManager

Bases: Manager

Manager for the LogCard model — atomic counter upserts.

upsert(*, logger_name, total_delta, error_delta, warning_delta, last_seen)

Create or atomically increment counters for logger_name.

Source code in log_panel/managers.py

def upsert(
    self,
    *,
    logger_name: str,
    total_delta: int,
    error_delta: int,
    warning_delta: int,
    last_seen: datetime,
) -> None:
    """Create or atomically increment counters for *logger_name*."""
    from log_panel.models import Logger

    logger_obj, _ = Logger.objects.db_manager(self.db).get_or_create(
        name=logger_name
    )
    _, created = self.get_or_create(
        logger=logger_obj,
        defaults={
            "total": total_delta,
            "total_errors": error_delta,
            "total_warnings": warning_delta,
            "last_seen": last_seen,
        },
    )
    if not created:
        updates: dict[str, Any] = {
            "total": F("total") + total_delta,
        }
        if error_delta:
            updates["total_errors"] = F("total_errors") + error_delta
        if warning_delta:
            updates["total_warnings"] = F("total_warnings") + warning_delta
        updates["last_seen"] = Greatest(F("last_seen"), last_seen)
        self.filter(logger=logger_obj).update(**updates)

replace_snapshot(*, logger_name, total, total_errors, total_warnings, last_seen)

Replace counters for logger_name with an exact rebuild snapshot.

Source code in log_panel/managers.py

def replace_snapshot(
    self,
    *,
    logger_name: str,
    total: int,
    total_errors: int,
    total_warnings: int,
    last_seen: datetime,
) -> None:
    """Replace counters for *logger_name* with an exact rebuild snapshot."""
    from log_panel.models import Logger

    logger_obj, _ = Logger.objects.db_manager(self.db).get_or_create(
        name=logger_name
    )
    self.update_or_create(
        logger=logger_obj,
        defaults={
            "total": total,
            "total_errors": total_errors,
            "total_warnings": total_warnings,
            "last_seen": last_seen,
        },
    )

TimelineBucketManager

Bases: Manager

Manager for the LogTimelineBucket model — atomic bucket upserts.

upsert(*, logger_name, timestamp, level)

Create or increment hourly and daily buckets for a single log record.

Source code in log_panel/managers.py

def upsert(self, *, logger_name: str, timestamp: datetime, level: str) -> None:
    """Create or increment hourly and daily buckets for a single log record."""
    from log_panel.models import Logger

    logger_obj, _ = Logger.objects.db_manager(self.db).get_or_create(
        name=logger_name
    )
    error_delta: Literal[1, 0] = 1 if level in ERROR_LEVELS else 0
    warning_delta: Literal[1, 0] = 1 if level == LogLevel.WARNING else 0

    hour_bucket = timestamp.replace(minute=0, second=0, microsecond=0)
    day_bucket = timestamp.replace(hour=0, minute=0, second=0, microsecond=0)

    for bucket, unit in (
        (hour_bucket, RangeUnit.HOUR),
        (day_bucket, RangeUnit.DAY),
    ):
        self._upsert_single(
            logger=logger_obj,
            bucket=bucket,
            unit=unit,
            log_count_delta=1,
            error_delta=error_delta,
            warning_delta=warning_delta,
        )

bulk_upsert(records)

Aggregate and upsert timeline buckets for a batch of log records.

Source code in log_panel/managers.py

def bulk_upsert(self, records: list[dict[str, Any]]) -> None:
    """Aggregate and upsert timeline buckets for a batch of log records."""
    from log_panel.models import Logger

    BucketKey = tuple[str, datetime, str]
    deltas: dict[BucketKey, list[int]] = {}

    for r in records:
        level = r["level"]
        error_delta: Literal[1, 0] = 1 if level in ERROR_LEVELS else 0
        warning_delta: Literal[1, 0] = 1 if level == LogLevel.WARNING else 0

        ts: datetime = to_database_datetime(value=r["timestamp"])
        name = r["logger_name"]

        hour_bucket = ts.replace(minute=0, second=0, microsecond=0)
        day_bucket = ts.replace(hour=0, minute=0, second=0, microsecond=0)

        for bucket, unit in (
            (hour_bucket, RangeUnit.HOUR),
            (day_bucket, RangeUnit.DAY),
        ):
            key: BucketKey = (name, bucket, unit)
            if key in deltas:
                deltas[key][0] += 1
                deltas[key][1] += error_delta
                deltas[key][2] += warning_delta
            else:
                deltas[key] = [1, error_delta, warning_delta]

    for (logger_name, bucket, unit), (lc, ed, wd) in deltas.items():
        logger_obj, _ = Logger.objects.db_manager(self.db).get_or_create(
            name=logger_name
        )
        self._upsert_single(
            logger=logger_obj,
            bucket=bucket,
            unit=unit,
            log_count_delta=lc,
            error_delta=ed,
            warning_delta=wd,
        )

replace_snapshot(*, logger_name, bucket, unit, log_count, error_count, warning_count)

Replace one timeline bucket with an exact rebuild snapshot.

Source code in log_panel/managers.py

def replace_snapshot(
    self,
    *,
    logger_name: str,
    bucket: datetime,
    unit: str,
    log_count: int,
    error_count: int,
    warning_count: int,
) -> None:
    """Replace one timeline bucket with an exact rebuild snapshot."""
    from log_panel.models import Logger

    logger_obj, _ = Logger.objects.db_manager(self.db).get_or_create(
        name=logger_name
    )
    self.update_or_create(
        logger=logger_obj,
        bucket=bucket,
        unit=unit,
        defaults={
            "log_count": log_count,
            "error_count": error_count,
            "warning_count": warning_count,
        },
    )