Spaces:
Runtime error
Runtime error
File size: 3,400 Bytes
decd9fb | 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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """
Lightweight in-process cache for high-frequency public reads.
"""
import asyncio
import time
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models import Notification, Session
_active_sessions_cache: list[dict[str, Any]] = []
_active_sessions_expires_at: float = 0.0
_active_sessions_lock = asyncio.Lock()
_recent_notifications_cache: list[dict[str, Any]] = []
_recent_notifications_expires_at: float = 0.0
_recent_notifications_lock = asyncio.Lock()
def _serialize_session(row: Session) -> dict[str, Any]:
return {
"id": row.id,
"session_name": row.session_name,
"start_time": row.start_time.isoformat() if row.start_time else None,
"end_time": row.end_time.isoformat() if row.end_time else None,
"status": row.status,
}
def _serialize_notification(row: Notification) -> dict[str, Any]:
return {
"id": row.id,
"message": row.message,
"priority": row.priority,
"created_by": row.created_by,
"created_at": row.created_at.isoformat() if row.created_at else None,
}
async def get_active_sessions_cached(
db: AsyncSession,
*,
force_refresh: bool = False,
) -> list[dict[str, Any]]:
global _active_sessions_cache, _active_sessions_expires_at
now = time.monotonic()
if not force_refresh and now < _active_sessions_expires_at:
return _active_sessions_cache
async with _active_sessions_lock:
now = time.monotonic()
if not force_refresh and now < _active_sessions_expires_at:
return _active_sessions_cache
rows = (
await db.execute(
select(Session)
.filter(Session.status == "active")
.order_by(Session.start_time.desc())
)
).scalars().all()
_active_sessions_cache = [_serialize_session(row) for row in rows]
_active_sessions_expires_at = now + float(settings.ACTIVE_SESSIONS_CACHE_TTL_SECONDS)
return _active_sessions_cache
async def get_recent_notifications_cached(
db: AsyncSession,
*,
force_refresh: bool = False,
) -> list[dict[str, Any]]:
global _recent_notifications_cache, _recent_notifications_expires_at
now = time.monotonic()
if not force_refresh and now < _recent_notifications_expires_at:
return _recent_notifications_cache
async with _recent_notifications_lock:
now = time.monotonic()
if not force_refresh and now < _recent_notifications_expires_at:
return _recent_notifications_cache
rows = (
await db.execute(
select(Notification)
.order_by(Notification.created_at.desc())
.limit(settings.RECENT_NOTIFICATIONS_LIMIT)
)
).scalars().all()
_recent_notifications_cache = [_serialize_notification(row) for row in rows]
_recent_notifications_expires_at = now + float(
settings.NOTIFICATIONS_CACHE_TTL_SECONDS
)
return _recent_notifications_cache
def invalidate_active_sessions_cache() -> None:
global _active_sessions_expires_at
_active_sessions_expires_at = 0.0
def invalidate_recent_notifications_cache() -> None:
global _recent_notifications_expires_at
_recent_notifications_expires_at = 0.0
|