90 lines
2.4 KiB
Python
90 lines
2.4 KiB
Python
"""
|
|
Simple in-memory cache for celestial data
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
import logging
|
|
|
|
from app.models.celestial import CelestialBody
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CacheEntry:
|
|
"""Cache entry with expiration"""
|
|
|
|
def __init__(self, data: list[CelestialBody], ttl_days: int = 3):
|
|
self.data = data
|
|
self.created_at = datetime.utcnow()
|
|
self.expires_at = self.created_at + timedelta(days=ttl_days)
|
|
|
|
def is_expired(self) -> bool:
|
|
"""Check if cache entry is expired"""
|
|
return datetime.utcnow() > self.expires_at
|
|
|
|
|
|
class CacheService:
|
|
"""Simple in-memory cache service"""
|
|
|
|
def __init__(self):
|
|
self._cache: dict[str, CacheEntry] = {}
|
|
|
|
def _make_key(
|
|
self,
|
|
start_time: datetime | None,
|
|
end_time: datetime | None,
|
|
step: str,
|
|
) -> str:
|
|
"""Generate cache key from query parameters"""
|
|
start_str = start_time.isoformat() if start_time else "now"
|
|
end_str = end_time.isoformat() if end_time else "now"
|
|
return f"{start_str}_{end_str}_{step}"
|
|
|
|
def get(
|
|
self,
|
|
start_time: datetime | None,
|
|
end_time: datetime | None,
|
|
step: str,
|
|
) -> Optional[list[CelestialBody]]:
|
|
"""
|
|
Get cached data if available and not expired
|
|
|
|
Returns:
|
|
Cached data or None if not found/expired
|
|
"""
|
|
key = self._make_key(start_time, end_time, step)
|
|
|
|
if key in self._cache:
|
|
entry = self._cache[key]
|
|
if not entry.is_expired():
|
|
logger.info(f"Cache hit for key: {key}")
|
|
return entry.data
|
|
else:
|
|
logger.info(f"Cache expired for key: {key}")
|
|
del self._cache[key]
|
|
|
|
logger.info(f"Cache miss for key: {key}")
|
|
return None
|
|
|
|
def set(
|
|
self,
|
|
data: list[CelestialBody],
|
|
start_time: datetime | None,
|
|
end_time: datetime | None,
|
|
step: str,
|
|
):
|
|
"""Store data in cache"""
|
|
key = self._make_key(start_time, end_time, step)
|
|
self._cache[key] = CacheEntry(data, ttl_days=settings.cache_ttl_days)
|
|
logger.info(f"Cached data for key: {key}")
|
|
|
|
def clear(self):
|
|
"""Clear all cache"""
|
|
self._cache.clear()
|
|
logger.info("Cache cleared")
|
|
|
|
|
|
# Singleton instance
|
|
cache_service = CacheService()
|