| import time |
| from collections import deque |
| from typing import Optional |
|
|
|
|
| class LogEntry: |
| __slots__ = ( |
| "timestamp", |
| "model", |
| "token_name", |
| "stream", |
| "status", |
| "duration", |
| "error", |
| ) |
|
|
| def __init__( |
| self, |
| model: str, |
| token_name: str, |
| stream: bool, |
| status: str = "pending", |
| duration: float = 0, |
| error: str = "", |
| ): |
| self.timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) |
| self.model = model |
| self.token_name = token_name |
| self.stream = stream |
| self.status = status |
| self.duration = round(duration, 2) |
| self.error = error |
|
|
| def to_dict(self) -> dict: |
| return { |
| "timestamp": self.timestamp, |
| "model": self.model, |
| "token_name": self.token_name, |
| "stream": self.stream, |
| "status": self.status, |
| "duration": self.duration, |
| "error": self.error, |
| } |
|
|
|
|
| class LogStore: |
| def __init__(self, max_entries: int = 500): |
| self._logs: deque[LogEntry] = deque(maxlen=max_entries) |
| self.total_requests: int = 0 |
| self.total_success: int = 0 |
| self.total_errors: int = 0 |
|
|
| def add(self, entry: LogEntry): |
| self._logs.appendleft(entry) |
| self.total_requests += 1 |
| if entry.status == "success": |
| self.total_success += 1 |
| elif entry.status == "error": |
| self.total_errors += 1 |
|
|
| def resize(self, max_entries: int): |
| old = list(self._logs) |
| self._logs = deque(old, maxlen=max_entries) |
|
|
| def query( |
| self, |
| status: Optional[str] = None, |
| page: int = 1, |
| page_size: int = 50, |
| ) -> dict: |
| items = list(self._logs) |
| if status: |
| items = [e for e in items if e.status == status] |
| total = len(items) |
| start = (page - 1) * page_size |
| end = start + page_size |
| return { |
| "total": total, |
| "page": page, |
| "page_size": page_size, |
| "items": [e.to_dict() for e in items[start:end]], |
| } |
|
|
