"""
In-memory user cache with TTL.
This caches user data from Laravel to avoid calling the API on every request.
When Redis is available, this can be easily swapped out.
"""
from __future__ import annotations
import time
from typing import Optional, Dict
from dataclasses import dataclass
import httpx
from src.config import settings
from src.observability import get_logger
logger = get_logger(__name__)
@dataclass
class CachedUser:
"""Cached user data with expiration."""
user_id: int
email: str
name: str
role_id: int
organization_id: int
platform_id: Optional[int]
dealership_id: Optional[int]
permissions: list[str]
user_info: Dict
user_organization: Dict
user_role: Dict
raw_data: Dict # Full raw response from Laravel (for tools)
cached_at: float # timestamp
ttl: int = 900 # 15 minutes default
def is_expired(self) -> bool:
"""Check if cache entry has expired."""
return (time.time() - self.cached_at) > self.ttl
class UserCache:
"""
Simple in-memory cache for user data.
This will be replaced with Redis in production, but works fine
for development and low-traffic scenarios.
"""
def __init__(self, ttl: int = 900):
"""
Initialize user cache.
Args:
ttl: Time to live in seconds (default 15 minutes)
"""
self._cache: Dict[str, CachedUser] = {}
self.ttl = ttl
self.hits = 0
self.misses = 0
def get(self, token: str) -> Optional[CachedUser]:
"""
Get cached user data by token.
Args:
token: Bearer token
Returns:
CachedUser if found and not expired, None otherwise
"""
cached = self._cache.get(token)
if cached is None:
self.misses += 1
logger.debug("cache_miss", token_prefix=token[:10])
return None
if cached.is_expired():
# Remove expired entry
del self._cache[token]
self.misses += 1
logger.debug("cache_expired", token_prefix=token[:10], user_id=cached.user_id)
return None
self.hits += 1
logger.debug(
"cache_hit",
token_prefix=token[:10],
user_id=cached.user_id,
age_seconds=int(time.time() - cached.cached_at)
)
return cached
def set(self, token: str, user_data: Dict) -> CachedUser:
"""
Cache user data.
Args:
token: Bearer token
user_data: User data from Laravel API
Returns:
CachedUser object
"""
# Extract platform_id and dealership_id from organization data
org = user_data.get("user_organization", {})
platform_id = None
dealership_id = None
# If organization has platforms, this might be platform-level
# You'll need to adjust this based on your actual data structure
if org.get("is_platform") == "y":
platform_id = org.get("id")
cached = CachedUser(
user_id=user_data["id"],
email=user_data["email"],
name=user_data["name"],
role_id=user_data["role"],
organization_id=user_data["organization"],
platform_id=platform_id,
dealership_id=dealership_id,
permissions=[p["name"] for p in user_data.get("user_permission", [])],
user_info=user_data.get("user_info", {}),
user_organization=user_data.get("user_organization", {}),
user_role=user_data.get("user_role", {}),
raw_data=user_data, # Store raw user payload for tools
cached_at=time.time(),
ttl=self.ttl
)
self._cache[token] = cached
logger.info(
"user_cached",
user_id=cached.user_id,
role_id=cached.role_id,
ttl=self.ttl
)
return cached
def invalidate(self, token: str) -> bool:
"""
Invalidate cached user data.
Args:
token: Bearer token
Returns:
True if entry was removed, False if not found
"""
if token in self._cache:
user_id = self._cache[token].user_id
del self._cache[token]
logger.info("cache_invalidated", token_prefix=token[:10], user_id=user_id)
return True
return False
def clear(self):
"""Clear all cached data."""
count = len(self._cache)
self._cache.clear()
logger.info("cache_cleared", entries_removed=count)
def get_stats(self) -> Dict:
"""Get cache statistics."""
total = self.hits + self.misses
hit_rate = (self.hits / total * 100) if total > 0 else 0
return {
"entries": len(self._cache),
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{hit_rate:.1f}%",
"ttl": self.ttl
}
async def fetch_user_from_laravel(token: str) -> Optional[Dict]:
"""
Fetch user data from Laravel API.
Args:
token: Bearer token
Returns:
User data dict or None if failed
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{settings.updation_api_base_url}/v2/get-login-user-data",
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json"
},
timeout=5.0
)
if response.status_code == 200:
data = response.json()
if data.get("success"):
logger.info(
"user_fetched_from_laravel",
user_id=data["data"]["user"]["id"],
role_id=data["data"]["user"]["role"]
)
return data["data"]["user"]
logger.warning(
"laravel_fetch_failed",
status_code=response.status_code,
response=response.text[:200]
)
return None
except Exception as e:
logger.error(
"laravel_fetch_error",
error=str(e),
error_type=type(e).__name__
)
return None
# Global cache instance
# In production, this would be Redis
_user_cache = UserCache(ttl=900) # 15 minutes
async def get_user_from_token(token: str) -> Optional[CachedUser]:
"""
Get user data from token (cached or fresh from Laravel).
This is the main function to use. It handles caching automatically.
Args:
token: Bearer token
Returns:
CachedUser if valid, None if invalid token
"""
# Try cache first
cached = _user_cache.get(token)
if cached:
logger.info("fetching_user_from_cache", user_id=cached.user_id)
return cached
# Cache miss - fetch from Laravel
logger.info("fetching_user_from_laravel", token_prefix=token[:10])
user_data = await fetch_user_from_laravel(token)
if not user_data:
return None
# Cache and return
return _user_cache.set(token, user_data)
def get_cache_stats() -> Dict:
"""Get cache statistics for monitoring."""
return _user_cache.get_stats()
def clear_cache():
"""Clear all cached user data."""
_user_cache.clear()
