jpmschweitzerandClaude Opus 4.5
72f515bf61
- POST /volatile/fetch/environment/{city} fetches both in parallel
- Single geocode lookup shared between API calls
- Uses asyncio.gather() for concurrent external requests
- Fix scheduler executor name (rest_api → rest_api_executor)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
656 lines
20 KiB
Python
656 lines
20 KiB
Python
"""
|
|
Volatile cache router for Library Desk API.
|
|
|
|
Endpoints for ephemeral cached data with TTL - weather, news, financial, etc.
|
|
Data is stored as vectors in Qdrant for semantic search retrieval.
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
import logging
|
|
|
|
from src.models.volatile import (
|
|
VolatileRecordCreate,
|
|
VolatileRecordResponse,
|
|
VolatileListResponse,
|
|
VolatileScheduledResponse,
|
|
VolatileStatsResponse,
|
|
VolatileDeleteResponse,
|
|
VolatileNamespace,
|
|
NAMESPACE_DEFAULT_TTL,
|
|
)
|
|
from src.services.volatile_service import VolatileCacheService
|
|
from src.services.volatile_fetch_service import VolatileFetchService
|
|
from src.core.dependencies import (
|
|
verify_api_key,
|
|
QdrantDep,
|
|
OllamaDep,
|
|
get_weather_provider,
|
|
get_news_provider,
|
|
get_alphavantage_provider,
|
|
)
|
|
from src.core.multi_tenancy import DEFAULT_USER
|
|
from src.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/volatile", tags=["Volatile Cache"])
|
|
|
|
|
|
def get_volatile_service(qdrant: QdrantDep, ollama: OllamaDep) -> VolatileCacheService:
|
|
"""Get volatile cache service instance."""
|
|
settings = get_settings()
|
|
return VolatileCacheService(
|
|
qdrant_client=qdrant,
|
|
ollama_client=ollama,
|
|
settings=settings
|
|
)
|
|
|
|
|
|
@router.get("/stats", response_model=VolatileStatsResponse)
|
|
async def get_stats(
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Get volatile cache statistics.
|
|
|
|
Returns counts of records by namespace and scheduled refresh info.
|
|
"""
|
|
service = get_volatile_service(qdrant, ollama)
|
|
stats = await service.get_stats(user)
|
|
|
|
return VolatileStatsResponse(
|
|
total_records=stats["total_records"],
|
|
by_namespace=stats["by_namespace"],
|
|
scheduled_count=stats["scheduled_count"],
|
|
total_memory_bytes=None,
|
|
user=user,
|
|
)
|
|
|
|
|
|
@router.get("/scheduled", response_model=VolatileScheduledResponse)
|
|
async def get_scheduled(
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Get records with refresh schedules.
|
|
|
|
Used by scheduler to determine what volatile data needs refreshing.
|
|
Returns all records that have a refresh_schedule cron expression set.
|
|
"""
|
|
service = get_volatile_service(qdrant, ollama)
|
|
records = await service.get_scheduled(user)
|
|
|
|
return VolatileScheduledResponse(
|
|
records=records,
|
|
count=len(records),
|
|
user=user,
|
|
)
|
|
|
|
|
|
@router.get("/namespaces")
|
|
async def list_namespaces(
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
List available namespaces and their default TTLs.
|
|
|
|
Returns predefined namespaces with their default TTL values.
|
|
"""
|
|
return {
|
|
"namespaces": [
|
|
{
|
|
"name": ns.value,
|
|
"default_ttl": NAMESPACE_DEFAULT_TTL.get(ns, 3600),
|
|
"description": _get_namespace_description(ns),
|
|
}
|
|
for ns in VolatileNamespace
|
|
]
|
|
}
|
|
|
|
|
|
def _get_namespace_description(ns: VolatileNamespace) -> str:
|
|
"""Get human-readable description for namespace."""
|
|
descriptions = {
|
|
VolatileNamespace.WEATHER: "Weather conditions and forecasts",
|
|
VolatileNamespace.NEWS: "Headlines and breaking news",
|
|
VolatileNamespace.FINANCIAL: "Stock prices, exchange rates, crypto",
|
|
VolatileNamespace.TRANSIT: "Train/bus schedules, delays",
|
|
VolatileNamespace.TRAFFIC: "Commute times, road conditions",
|
|
VolatileNamespace.AIR_QUALITY: "Pollution levels, pollen counts",
|
|
VolatileNamespace.SPORTS: "Live scores, upcoming matches",
|
|
VolatileNamespace.SOCIAL: "Social media mentions, notifications",
|
|
VolatileNamespace.SYSTEM: "Service health, infrastructure status",
|
|
VolatileNamespace.CONTEXT: "Conversation context, session state",
|
|
VolatileNamespace.CUSTOM: "User-defined volatile data",
|
|
}
|
|
return descriptions.get(ns, "Custom namespace")
|
|
|
|
|
|
@router.get("/search")
|
|
async def search_volatile(
|
|
q: str = Query(..., min_length=1, description="Search query"),
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
limit: int = Query(default=5, ge=1, le=20, description="Maximum results"),
|
|
threshold: float = Query(default=0.75, ge=0.5, le=1.0, description="Minimum similarity score"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Semantic search across volatile data.
|
|
|
|
Searches all volatile data for semantically similar content.
|
|
Higher threshold = stricter matching.
|
|
|
|
**Example:**
|
|
```
|
|
GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer
|
|
```
|
|
"""
|
|
service = get_volatile_service(qdrant, ollama)
|
|
results = await service.search(user, q, limit=limit, score_threshold=threshold)
|
|
|
|
return {
|
|
"query": q,
|
|
"results": results,
|
|
"count": len(results),
|
|
"user": user,
|
|
}
|
|
|
|
|
|
@router.post("/store", response_model=VolatileRecordResponse)
|
|
async def store_volatile(
|
|
namespace: str = Query(..., description="Data namespace (weather, news, etc.)"),
|
|
key: str = Query(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')"),
|
|
request: VolatileRecordCreate = None,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Store volatile data.
|
|
|
|
Data is converted to natural language and embedded for semantic search.
|
|
If the same namespace+key already exists, it will be updated.
|
|
|
|
**Example Request:**
|
|
```json
|
|
POST /volatile/store?namespace=weather&key=rotterdam
|
|
{
|
|
"data": {
|
|
"temperature": 8,
|
|
"conditions": "Cloudy",
|
|
"humidity": 85
|
|
},
|
|
"source": "openweathermap",
|
|
"ttl": 1800,
|
|
"refresh_schedule": "0 * * * *"
|
|
}
|
|
```
|
|
|
|
**Refresh Schedule:**
|
|
Optional cron expression for automatic refresh. The scheduler
|
|
will query `/volatile/scheduled` and trigger refreshes.
|
|
"""
|
|
# Validate namespace if not custom
|
|
if namespace != VolatileNamespace.CUSTOM:
|
|
try:
|
|
VolatileNamespace(namespace)
|
|
except ValueError:
|
|
valid = [ns.value for ns in VolatileNamespace]
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid namespace '{namespace}'. Valid: {valid}"
|
|
)
|
|
|
|
service = get_volatile_service(qdrant, ollama)
|
|
|
|
try:
|
|
record = await service.store(
|
|
user=user,
|
|
namespace=namespace,
|
|
key=key,
|
|
data=request.data,
|
|
source=request.source,
|
|
ttl=request.ttl,
|
|
refresh_schedule=request.refresh_schedule,
|
|
)
|
|
return record
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to store volatile record: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to store record: {str(e)}")
|
|
|
|
|
|
@router.post("/fetch/weather/{city}")
|
|
async def fetch_weather(
|
|
city: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
ttl: int = Query(default=3600, ge=60, le=86400, description="TTL in seconds (default 1 hour)"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch current weather conditions for a city and store in volatile cache.
|
|
|
|
Stores temperature, humidity, wind, UV index. For forecasts use /fetch/forecast.
|
|
Called by scheduler for hourly prefetch or on-demand.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/weather/amsterdam?user=jpmschweitzer
|
|
```
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_current_weather(user, city, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/forecast/{city}")
|
|
async def fetch_forecast(
|
|
city: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
days: int = Query(default=7, ge=1, le=16, description="Forecast days (1-16)"),
|
|
ttl: int = Query(default=43200, ge=60, le=604800, description="TTL in seconds (default 12 hours)"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch weather forecast for a city and store in volatile cache.
|
|
|
|
Stores multi-day outlook with highs/lows, precipitation, UV.
|
|
For current conditions use /fetch/weather.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/forecast/amsterdam?user=jpmschweitzer&days=7
|
|
```
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_forecast(user, city, days=days, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/news/{category}")
|
|
async def fetch_news(
|
|
category: str = "general",
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
limit: int = Query(default=10, ge=1, le=50, description="Max headlines"),
|
|
ttl: int = Query(default=7200, ge=60, le=86400, description="TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch news headlines and store in volatile cache.
|
|
|
|
Fetches from configured news sources (NOS, BBC) based on user settings.
|
|
Categories: general, world, tech, business, politics, etc.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/news/tech?user=jpmschweitzer&limit=15
|
|
```
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
news_provider = await get_news_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
news_provider=news_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_news(user, category, limit=limit, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/stock/{symbol}")
|
|
async def fetch_stock(
|
|
symbol: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
ttl: int = Query(default=300, ge=60, le=3600, description="TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch stock quote and store in volatile cache.
|
|
|
|
Fetches from Alpha Vantage API. Requires API key configured in settings.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/stock/AAPL?user=jpmschweitzer
|
|
```
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
financial_provider = await get_alphavantage_provider()
|
|
|
|
if not financial_provider:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Financial provider not configured (Alpha Vantage API key missing)"
|
|
)
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
financial_provider=financial_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_stock(user, symbol, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/crypto/{symbol}")
|
|
async def fetch_crypto(
|
|
symbol: str,
|
|
market: str = Query(default="USD", description="Market currency"),
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
ttl: int = Query(default=300, ge=60, le=3600, description="TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch cryptocurrency quote and store in volatile cache.
|
|
|
|
Fetches from Alpha Vantage API. Requires API key configured in settings.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/crypto/BTC?market=EUR&user=jpmschweitzer
|
|
```
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
financial_provider = await get_alphavantage_provider()
|
|
|
|
if not financial_provider:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Financial provider not configured (Alpha Vantage API key missing)"
|
|
)
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
financial_provider=financial_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_crypto(user, symbol, market=market, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/sun/{city}")
|
|
async def fetch_sun_times(
|
|
city: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
ttl: int = Query(default=86400, ge=60, le=604800, description="TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch sunrise/sunset times for a city and store in volatile cache.
|
|
|
|
Fetches from Open-Meteo API. Useful for home automation triggers.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/sun/rotterdam?user=jpmschweitzer
|
|
```
|
|
|
|
**Response data includes:**
|
|
- sunrise/sunset times (both HH:MM and ISO formats)
|
|
- daylight_duration_seconds
|
|
- daylight_hours
|
|
- Natural language text summary
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_sun_times(user, city, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/air_quality/{city}")
|
|
async def fetch_air_quality(
|
|
city: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
ttl: int = Query(default=3600, ge=60, le=86400, description="TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch air quality data for a city and store in volatile cache.
|
|
|
|
Fetches from Open-Meteo Air Quality API.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/air_quality/rotterdam?user=jpmschweitzer
|
|
```
|
|
|
|
**Response data includes:**
|
|
- European and US AQI indices
|
|
- Pollutants: PM2.5, PM10, ozone, nitrogen dioxide, etc.
|
|
- Pollen data (European locations, seasonal)
|
|
- Natural language text summary
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_air_quality(user, city, ttl=ttl)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
|
|
return {
|
|
"success": True,
|
|
"namespace": result.namespace,
|
|
"key": result.key,
|
|
"record": result.record,
|
|
}
|
|
|
|
|
|
@router.post("/fetch/environment/{city}")
|
|
async def fetch_environment(
|
|
city: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
weather_ttl: int = Query(default=3600, ge=60, le=86400, description="Weather TTL in seconds"),
|
|
air_quality_ttl: int = Query(default=3600, ge=60, le=86400, description="Air quality TTL in seconds"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Fetch weather and air quality concurrently for a city.
|
|
|
|
Performs a single geocode lookup and fetches both weather and air quality
|
|
data in parallel, storing both in volatile cache. More efficient than
|
|
calling /fetch/weather and /fetch/air_quality separately.
|
|
|
|
**Example:**
|
|
```
|
|
POST /volatile/fetch/environment/rotterdam?user=jpmschweitzer
|
|
```
|
|
|
|
**Response includes:**
|
|
- weather: Current conditions (temperature, humidity, wind, UV)
|
|
- air_quality: AQI indices, pollutants, pollen data
|
|
"""
|
|
volatile_service = get_volatile_service(qdrant, ollama)
|
|
weather_provider = get_weather_provider()
|
|
|
|
fetch_service = VolatileFetchService(
|
|
volatile_service=volatile_service,
|
|
weather_provider=weather_provider,
|
|
)
|
|
|
|
result = await fetch_service.fetch_environment(
|
|
user, city, weather_ttl=weather_ttl, air_quality_ttl=air_quality_ttl
|
|
)
|
|
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail="; ".join(result.errors))
|
|
|
|
return {
|
|
"success": True,
|
|
"key": result.key,
|
|
"weather": {
|
|
"success": result.weather.success if result.weather else False,
|
|
"record": result.weather.record if result.weather else None,
|
|
"error": result.weather.error if result.weather else None,
|
|
},
|
|
"air_quality": {
|
|
"success": result.air_quality.success if result.air_quality else False,
|
|
"record": result.air_quality.record if result.air_quality else None,
|
|
"error": result.air_quality.error if result.air_quality else None,
|
|
},
|
|
"errors": result.errors,
|
|
}
|
|
|
|
|
|
@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse)
|
|
async def get_record(
|
|
namespace: str,
|
|
key: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Get a specific volatile record by namespace and key.
|
|
|
|
**Example:**
|
|
```
|
|
GET /volatile/weather/rotterdam?user=jpmschweitzer
|
|
```
|
|
"""
|
|
service = get_volatile_service(qdrant, ollama)
|
|
record = await service.get(user, namespace, key)
|
|
|
|
if not record:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Record '{key}' not found in namespace '{namespace}'"
|
|
)
|
|
|
|
return record
|
|
|
|
|
|
@router.delete("/{namespace}/{key}", response_model=VolatileDeleteResponse)
|
|
async def delete_record(
|
|
namespace: str,
|
|
key: str,
|
|
user: str = Query(default=DEFAULT_USER, description="User identifier"),
|
|
qdrant: QdrantDep = None,
|
|
ollama: OllamaDep = None,
|
|
api_key: str = Depends(verify_api_key)
|
|
):
|
|
"""
|
|
Delete a specific volatile record.
|
|
"""
|
|
service = get_volatile_service(qdrant, ollama)
|
|
deleted = await service.delete(user, namespace, key)
|
|
|
|
return VolatileDeleteResponse(
|
|
key=key,
|
|
namespace=namespace,
|
|
deleted=deleted,
|
|
user=user,
|
|
)
|