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) -> LogQueryset:
    """Returns a LogQueryset for the active backend with no filters applied."""
    from log_panel import conf

    return LogQueryset(backend=conf.get_backend())

LogRecordManager

Bases: Manager

Custom manager for Log

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,
        )
    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 LogMessageChunk

    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
        )

    return created_logs