jpmschweitzerandClaude Opus 4.5
4cfad2e8bc
TTL = 2x refresh interval ensures data remains valid even if a scheduled refresh is delayed or fails. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
136 lines
6.6 KiB
Python
136 lines
6.6 KiB
Python
"""
|
|
Volatile memory models for Library Desk.
|
|
|
|
Provides models for ephemeral cached data with TTL - weather, news, financial data,
|
|
transit schedules, and other time-sensitive external information.
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field
|
|
from typing import Dict, Any, Optional, List
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
|
|
class VolatileNamespace(str, Enum):
|
|
"""
|
|
Predefined namespaces for volatile data.
|
|
|
|
Each namespace can have different default TTLs and refresh schedules.
|
|
"""
|
|
# Real-time external data
|
|
WEATHER = "weather" # Current conditions (temperature, humidity, wind)
|
|
FORECAST = "forecast" # Multi-day weather outlook
|
|
SUN = "sun" # Sunrise, sunset, daylight duration
|
|
NEWS = "news" # Headlines, breaking news
|
|
FINANCIAL = "financial" # Stock prices, exchange rates, crypto
|
|
TRANSIT = "transit" # Train/bus schedules, delays, disruptions
|
|
TRAFFIC = "traffic" # Commute times, road conditions
|
|
AIR_QUALITY = "air_quality" # Pollution levels, pollen counts
|
|
SPORTS = "sports" # Live scores, upcoming matches
|
|
|
|
# System/integration data
|
|
SOCIAL = "social" # Social media mentions, notifications
|
|
SYSTEM = "system" # Service health, infrastructure status
|
|
|
|
# Ephemeral context
|
|
CONTEXT = "context" # Conversation context, session state
|
|
CUSTOM = "custom" # User-defined volatile data
|
|
|
|
|
|
# Default TTLs per namespace (in seconds)
|
|
# TTL = 2x refresh interval to ensure data survives missed/delayed refreshes
|
|
NAMESPACE_DEFAULT_TTL: Dict[str, int] = {
|
|
VolatileNamespace.WEATHER: 7200, # 2 hours (hourly refresh)
|
|
VolatileNamespace.FORECAST: 86400, # 24 hours (12hr refresh)
|
|
VolatileNamespace.SUN: 172800, # 48 hours (daily refresh)
|
|
VolatileNamespace.NEWS: 7200, # 2 hours (hourly refresh)
|
|
VolatileNamespace.FINANCIAL: 600, # 10 min (5 min refresh)
|
|
VolatileNamespace.TRANSIT: 600, # 10 min (5 min refresh)
|
|
VolatileNamespace.TRAFFIC: 1200, # 20 min (10 min refresh)
|
|
VolatileNamespace.AIR_QUALITY: 7200, # 2 hours (hourly refresh)
|
|
VolatileNamespace.SPORTS: 120, # 2 min (1 min refresh)
|
|
VolatileNamespace.SOCIAL: 1200, # 20 min (10 min refresh)
|
|
VolatileNamespace.SYSTEM: 120, # 2 min (1 min refresh)
|
|
VolatileNamespace.CONTEXT: 7200, # 2 hours - session context
|
|
VolatileNamespace.CUSTOM: 7200, # 2 hours - default for custom
|
|
}
|
|
|
|
|
|
class VolatileRecord(BaseModel):
|
|
"""
|
|
A volatile cache record with TTL.
|
|
|
|
Volatile records are ephemeral data stored in Redis with automatic expiration.
|
|
Used for weather, news, financial data, and other time-sensitive information.
|
|
"""
|
|
key: str = Field(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')")
|
|
namespace: str = Field(..., description="Namespace (e.g., 'weather', 'news', 'financial')")
|
|
data: Dict[str, Any] = Field(..., description="Actual content/payload")
|
|
source: Optional[str] = Field(None, description="Origin API/service (e.g., 'openweathermap', 'nos.nl')")
|
|
created_at: datetime = Field(default_factory=datetime.utcnow, description="When record was created")
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow, description="When record was last updated")
|
|
ttl: int = Field(..., ge=60, le=604800, description="Time-to-live in seconds (max 7 days)")
|
|
refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh")
|
|
user: str = Field(..., description="User identifier for multi-tenancy")
|
|
|
|
|
|
class VolatileRecordCreate(BaseModel):
|
|
"""Request model for creating/updating a volatile record."""
|
|
data: Dict[str, Any] = Field(..., description="Content to store")
|
|
source: Optional[str] = Field(None, description="Origin API/service")
|
|
ttl: Optional[int] = Field(None, ge=60, le=604800, description="TTL in seconds (uses namespace default if not set)")
|
|
refresh_schedule: Optional[str] = Field(None, description="Cron expression for scheduled refresh")
|
|
|
|
|
|
class VolatileRecordResponse(BaseModel):
|
|
"""Response model for a volatile record."""
|
|
key: str = Field(..., description="Record key")
|
|
namespace: str = Field(..., description="Namespace")
|
|
data: Dict[str, Any] = Field(..., description="Stored content")
|
|
source: Optional[str] = Field(None, description="Origin API/service")
|
|
created_at: datetime = Field(..., description="Creation timestamp")
|
|
updated_at: datetime = Field(..., description="Last update timestamp")
|
|
ttl: int = Field(..., description="TTL in seconds")
|
|
ttl_remaining: int = Field(..., description="Seconds until expiration")
|
|
refresh_schedule: Optional[str] = Field(None, description="Cron expression if scheduled")
|
|
user: str = Field(..., description="User identifier")
|
|
|
|
|
|
class VolatileListResponse(BaseModel):
|
|
"""Response model for listing volatile records."""
|
|
namespace: str = Field(..., description="Namespace queried")
|
|
keys: List[str] = Field(..., description="List of keys in namespace")
|
|
count: int = Field(..., description="Number of keys")
|
|
user: str = Field(..., description="User identifier")
|
|
|
|
|
|
class VolatileScheduledResponse(BaseModel):
|
|
"""Response model for records needing refresh."""
|
|
records: List[VolatileRecordResponse] = Field(..., description="Records with refresh schedules")
|
|
count: int = Field(..., description="Number of scheduled records")
|
|
user: str = Field(..., description="User identifier")
|
|
|
|
|
|
class VolatileStatsResponse(BaseModel):
|
|
"""Response model for volatile cache statistics."""
|
|
total_records: int = Field(..., description="Total volatile records for user")
|
|
by_namespace: Dict[str, int] = Field(..., description="Record count per namespace")
|
|
scheduled_count: int = Field(..., description="Records with refresh schedules")
|
|
total_memory_bytes: Optional[int] = Field(None, description="Approximate memory usage")
|
|
user: str = Field(..., description="User identifier")
|
|
|
|
|
|
class VolatileDeleteResponse(BaseModel):
|
|
"""Response model for delete operation."""
|
|
key: str = Field(..., description="Deleted key")
|
|
namespace: str = Field(..., description="Namespace")
|
|
deleted: bool = Field(..., description="Whether record was found and deleted")
|
|
user: str = Field(..., description="User identifier")
|
|
|
|
|
|
class VolatileBulkDeleteResponse(BaseModel):
|
|
"""Response model for bulk delete operations."""
|
|
namespace: Optional[str] = Field(None, description="Namespace if namespace-wide delete")
|
|
deleted_count: int = Field(..., description="Number of records deleted")
|
|
user: str = Field(..., description="User identifier")
|