Spaces:
Build error
Build error
File size: 1,238 Bytes
71b4454 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Any
logger = logging.getLogger("monitoring.telemetry")
class MonitoringEngine:
def __init__(self) -> None:
self.events: list[dict[str, Any]] = []
async def record_event(
self,
event: str,
project_id: str | None = None,
status: str = "info",
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
record = {
"event": event,
"project_id": project_id,
"status": status,
"metadata": metadata or {},
"timestamp": datetime.now(timezone.utc).isoformat(),
}
self.events.append(record)
if len(self.events) > 1000:
del self.events[:-1000]
logger.info("telemetry=%s", record)
return record
async def get_events(
self,
project_id: str | None = None,
) -> list[dict[str, Any]]:
if project_id is None:
return list(self.events)
return [
event
for event in self.events
if event.get("project_id") == project_id
]
monitoring_engine = MonitoringEngine()
|