File size: 1,436 Bytes
4035248 9f555a3 5bbddc0 9f555a3 4035248 |
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 |
from __future__ import annotations
import logging
from datetime import datetime
from typing import Callable, Optional
from zoneinfo import ZoneInfo
from motor.motor_asyncio import AsyncIOMotorCollection
logger = logging.getLogger(__name__)
class ApiLogger:
"""Persists API invocation metadata for observability/auditing."""
def __init__(self, collection_getter: Callable[[], AsyncIOMotorCollection]) -> None:
self._collection_getter = collection_getter
def _collection(self) -> AsyncIOMotorCollection:
return self._collection_getter()
@staticmethod
def _current_ist_timestamp() -> str:
now = datetime.now(ZoneInfo("Asia/Kolkata"))
return now.strftime("%d-%m-%Y %H:%M:%S:%Z")
async def log_categorization(
self,
*,
name: str,
status: str,
response_time: float,
user_id: Optional[str],
error_message: Optional[str] = None,
) -> None:
doc = {
"name": name,
"status": status,
"date": self._current_ist_timestamp(),
"response_time": round(response_time, 3),
}
if user_id:
doc["user_id"] = user_id
if error_message:
doc["error_message"] = error_message
try:
await self._collection().insert_one(doc)
except Exception:
logger.exception("Failed to write categorize API log")
|