Add Upstash Redis caching support and update dependencies

.env.example

@@ -50,3 +50,7 @@ CIRCUIT_BREAKER_RECOVERY_TIMEOUT=30
 # Cost Control
 MAX_TOKENS_PER_REQUEST=50000
 MONTHLY_TOKEN_BUDGET=10000000
+
+# Upstash Redis Configuration (for caching observability metrics)
+UPSTASH_REDIS_REST_URL=https://your-redis-instance.upstash.io
+UPSTASH_REDIS_REST_TOKEN=your_upstash_redis_token_here

app/core/cache.py

@@ -0,0 +1,469 @@
+"""
+Upstash Redis Cache Service for Observability Dashboard.
+
+Provides caching layer for LangSmith metrics with TTL support.
+Uses HTTP-based Redis client optimized for serverless environments.
+"""
+
+import os
+import json
+import logging
+from typing import Optional, Any, Dict, Union
+from datetime import datetime, timezone
+from functools import wraps
+from dataclasses import dataclass, asdict
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CacheConfig:
+    """Cache configuration with sensible defaults."""
+    # TTL values in seconds
+    summary_ttl: int = 300          # 5 minutes for summary stats
+    trends_ttl: int = 600           # 10 minutes for trend data
+    projects_ttl: int = 300         # 5 minutes for project list
+    runs_ttl: int = 120             # 2 minutes for run details
+    default_ttl: int = 300          # 5 minutes default
+    
+    # Cache key prefixes
+    prefix: str = "ideasprinter:observability"
+
+
+class UpstashRedisCache:
+    """
+    Async-compatible Redis cache using Upstash REST API.
+    
+    Designed for serverless environments with HTTP-based connections.
+    Supports both sync and async operations.
+    """
+    
+    def __init__(self, config: Optional[CacheConfig] = None):
+        self.config = config or CacheConfig()
+        self._sync_client = None
+        self._async_client = None
+        self._initialized = False
+        
+        # Load credentials from environment
+        self.url = os.getenv("UPSTASH_REDIS_REST_URL")
+        self.token = os.getenv("UPSTASH_REDIS_REST_TOKEN")
+        
+        if not self.url or not self.token:
+            logger.warning(
+                "Upstash Redis credentials not found. "
+                "Set UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN. "
+                "Cache operations will be no-ops."
+            )
+    
+    @property
+    def is_configured(self) -> bool:
+        """Check if Redis is properly configured."""
+        return bool(self.url and self.token)
+    
+    def _get_sync_client(self):
+        """Get or create synchronous Redis client."""
+        if not self.is_configured:
+            return None
+            
+        if self._sync_client is None:
+            try:
+                from upstash_redis import Redis
+                self._sync_client = Redis(url=self.url, token=self.token)
+                logger.info("Upstash Redis sync client initialized")
+            except Exception as e:
+                logger.error(f"Failed to initialize Upstash Redis sync client: {e}")
+                return None
+        return self._sync_client
+    
+    def _get_async_client(self):
+        """Get or create asynchronous Redis client."""
+        if not self.is_configured:
+            return None
+            
+        if self._async_client is None:
+            try:
+                from upstash_redis.asyncio import Redis
+                self._async_client = Redis(url=self.url, token=self.token)
+                logger.info("Upstash Redis async client initialized")
+            except Exception as e:
+                logger.error(f"Failed to initialize Upstash Redis async client: {e}")
+                return None
+        return self._async_client
+    
+    def _make_key(self, key: str) -> str:
+        """Generate full cache key with prefix."""
+        return f"{self.config.prefix}:{key}"
+    
+    def _serialize(self, value: Any) -> str:
+        """Serialize value to JSON string."""
+        if isinstance(value, (dict, list)):
+            return json.dumps(value, default=str)
+        return json.dumps({"value": value}, default=str)
+    
+    def _deserialize(self, value: Optional[str]) -> Any:
+        """Deserialize JSON string to Python object."""
+        if value is None:
+            return None
+        try:
+            data = json.loads(value)
+            # Unwrap simple values
+            if isinstance(data, dict) and "value" in data and len(data) == 1:
+                return data["value"]
+            return data
+        except (json.JSONDecodeError, TypeError):
+            return value
+    
+    # ==================== Synchronous Operations ====================
+    
+    def get(self, key: str) -> Optional[Any]:
+        """Get value from cache (sync)."""
+        client = self._get_sync_client()
+        if not client:
+            return None
+        
+        try:
+            full_key = self._make_key(key)
+            value = client.get(full_key)
+            if value:
+                logger.debug(f"Cache HIT: {key}")
+                return self._deserialize(value)
+            logger.debug(f"Cache MISS: {key}")
+            return None
+        except Exception as e:
+            logger.warning(f"Cache get error for {key}: {e}")
+            return None
+    
+    def set(
+        self, 
+        key: str, 
+        value: Any, 
+        ttl: Optional[int] = None
+    ) -> bool:
+        """Set value in cache with TTL (sync)."""
+        client = self._get_sync_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            serialized = self._serialize(value)
+            ex = ttl or self.config.default_ttl
+            
+            result = client.set(full_key, serialized, ex=ex)
+            logger.debug(f"Cache SET: {key} (TTL: {ex}s)")
+            return bool(result)
+        except Exception as e:
+            logger.warning(f"Cache set error for {key}: {e}")
+            return False
+    
+    def delete(self, key: str) -> bool:
+        """Delete key from cache (sync)."""
+        client = self._get_sync_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            result = client.delete(full_key)
+            logger.debug(f"Cache DELETE: {key}")
+            return bool(result)
+        except Exception as e:
+            logger.warning(f"Cache delete error for {key}: {e}")
+            return False
+    
+    def exists(self, key: str) -> bool:
+        """Check if key exists in cache (sync)."""
+        client = self._get_sync_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            return bool(client.exists(full_key))
+        except Exception as e:
+            logger.warning(f"Cache exists error for {key}: {e}")
+            return False
+    
+    # ==================== Asynchronous Operations ====================
+    
+    async def aget(self, key: str) -> Optional[Any]:
+        """Get value from cache (async)."""
+        client = self._get_async_client()
+        if not client:
+            return None
+        
+        try:
+            full_key = self._make_key(key)
+            value = await client.get(full_key)
+            if value:
+                logger.debug(f"Cache HIT: {key}")
+                return self._deserialize(value)
+            logger.debug(f"Cache MISS: {key}")
+            return None
+        except Exception as e:
+            logger.warning(f"Async cache get error for {key}: {e}")
+            return None
+    
+    async def aset(
+        self, 
+        key: str, 
+        value: Any, 
+        ttl: Optional[int] = None
+    ) -> bool:
+        """Set value in cache with TTL (async)."""
+        client = self._get_async_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            serialized = self._serialize(value)
+            ex = ttl or self.config.default_ttl
+            
+            result = await client.set(full_key, serialized, ex=ex)
+            logger.debug(f"Async cache SET: {key} (TTL: {ex}s)")
+            return bool(result)
+        except Exception as e:
+            logger.warning(f"Async cache set error for {key}: {e}")
+            return False
+    
+    async def adelete(self, key: str) -> bool:
+        """Delete key from cache (async)."""
+        client = self._get_async_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            result = await client.delete(full_key)
+            logger.debug(f"Async cache DELETE: {key}")
+            return bool(result)
+        except Exception as e:
+            logger.warning(f"Async cache delete error for {key}: {e}")
+            return False
+    
+    async def aexists(self, key: str) -> bool:
+        """Check if key exists in cache (async)."""
+        client = self._get_async_client()
+        if not client:
+            return False
+        
+        try:
+            full_key = self._make_key(key)
+            return bool(await client.exists(full_key))
+        except Exception as e:
+            logger.warning(f"Async cache exists error for {key}: {e}")
+            return False
+    
+    # ==================== Observability-Specific Methods ====================
+    
+    def get_summary_key(self, user_id: str, start_date: str, end_date: str, project_id: Optional[str] = None) -> str:
+        """Generate cache key for summary stats."""
+        parts = ["summary", user_id, start_date, end_date]
+        if project_id:
+            parts.append(project_id)
+        return ":".join(parts)
+    
+    def get_trends_key(self, user_id: str, period: str, project_id: Optional[str] = None) -> str:
+        """Generate cache key for trends data."""
+        parts = ["trends", user_id, period]
+        if project_id:
+            parts.append(project_id)
+        return ":".join(parts)
+    
+    def get_projects_key(self, user_id: str) -> str:
+        """Generate cache key for projects list."""
+        return f"projects:{user_id}"
+    
+    def get_runs_key(self, user_id: str, project_id: str, page: int = 1) -> str:
+        """Generate cache key for runs list."""
+        return f"runs:{user_id}:{project_id}:page{page}"
+    
+    async def cache_summary(
+        self, 
+        user_id: str, 
+        start_date: str, 
+        end_date: str, 
+        data: Dict[str, Any],
+        project_id: Optional[str] = None
+    ) -> bool:
+        """Cache summary statistics."""
+        key = self.get_summary_key(user_id, start_date, end_date, project_id)
+        return await self.aset(key, data, ttl=self.config.summary_ttl)
+    
+    async def get_cached_summary(
+        self, 
+        user_id: str, 
+        start_date: str, 
+        end_date: str,
+        project_id: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """Get cached summary statistics."""
+        key = self.get_summary_key(user_id, start_date, end_date, project_id)
+        return await self.aget(key)
+    
+    async def cache_trends(
+        self, 
+        user_id: str, 
+        period: str, 
+        data: Dict[str, Any],
+        project_id: Optional[str] = None
+    ) -> bool:
+        """Cache trends data."""
+        key = self.get_trends_key(user_id, period, project_id)
+        return await self.aset(key, data, ttl=self.config.trends_ttl)
+    
+    async def get_cached_trends(
+        self, 
+        user_id: str, 
+        period: str,
+        project_id: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """Get cached trends data."""
+        key = self.get_trends_key(user_id, period, project_id)
+        return await self.aget(key)
+    
+    async def cache_projects(self, user_id: str, data: Dict[str, Any]) -> bool:
+        """Cache projects list."""
+        key = self.get_projects_key(user_id)
+        return await self.aset(key, data, ttl=self.config.projects_ttl)
+    
+    async def get_cached_projects(self, user_id: str) -> Optional[Dict[str, Any]]:
+        """Get cached projects list."""
+        key = self.get_projects_key(user_id)
+        return await self.aget(key)
+    
+    async def invalidate_user_cache(self, user_id: str) -> None:
+        """Invalidate all cache entries for a user."""
+        # Note: Upstash doesn't support SCAN easily via REST,
+        # so we invalidate known keys
+        patterns = [
+            f"summary:{user_id}:*",
+            f"trends:{user_id}:*",
+            f"projects:{user_id}",
+            f"runs:{user_id}:*"
+        ]
+        logger.info(f"Cache invalidation requested for user {user_id}")
+        # In practice, rely on TTL expiration for now
+        # For manual invalidation, track specific keys
+
+
+# ==================== Caching Decorator ====================
+
+def cached(
+    key_func: callable,
+    ttl: Optional[int] = None,
+    cache_instance: Optional[UpstashRedisCache] = None
+):
+    """
+    Decorator for caching async function results.
+    
+    Args:
+        key_func: Function that generates cache key from function arguments
+        ttl: Time-to-live in seconds (uses default if None)
+        cache_instance: Cache instance to use (creates default if None)
+    
+    Example:
+        @cached(
+            key_func=lambda user_id, period: f"trends:{user_id}:{period}",
+            ttl=600
+        )
+        async def get_trends(user_id: str, period: str):
+            # expensive operation
+            return await fetch_from_langsmith(...)
+    """
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            cache = cache_instance or get_cache()
+            
+            # Generate cache key
+            try:
+                cache_key = key_func(*args, **kwargs)
+            except Exception as e:
+                logger.warning(f"Failed to generate cache key: {e}")
+                return await func(*args, **kwargs)
+            
+            # Try to get from cache
+            cached_value = await cache.aget(cache_key)
+            if cached_value is not None:
+                return cached_value
+            
+            # Execute function and cache result
+            result = await func(*args, **kwargs)
+            
+            if result is not None:
+                await cache.aset(cache_key, result, ttl=ttl)
+            
+            return result
+        return wrapper
+    return decorator
+
+
+# ==================== Global Cache Instance ====================
+
+_cache_instance: Optional[UpstashRedisCache] = None
+
+
+def get_cache() -> UpstashRedisCache:
+    """Get or create the global cache instance."""
+    global _cache_instance
+    if _cache_instance is None:
+        _cache_instance = UpstashRedisCache()
+    return _cache_instance
+
+
+def init_cache(config: Optional[CacheConfig] = None) -> UpstashRedisCache:
+    """Initialize the global cache with custom config."""
+    global _cache_instance
+    _cache_instance = UpstashRedisCache(config)
+    return _cache_instance
+
+
+# ==================== Health Check ====================
+
+async def check_cache_health() -> Dict[str, Any]:
+    """
+    Check Redis cache health and connectivity.
+    
+    Returns:
+        Dict with status, latency, and configuration info
+    """
+    cache = get_cache()
+    result = {
+        "service": "upstash_redis",
+        "configured": cache.is_configured,
+        "status": "unknown",
+        "latency_ms": None,
+        "error": None
+    }
+    
+    if not cache.is_configured:
+        result["status"] = "not_configured"
+        result["error"] = "Missing UPSTASH_REDIS_REST_URL or UPSTASH_REDIS_REST_TOKEN"
+        return result
+    
+    try:
+        import time
+        start = time.time()
+        
+        # Ping test with a simple set/get
+        test_key = "_health_check"
+        await cache.aset(test_key, {"timestamp": datetime.now(timezone.utc).isoformat()}, ttl=10)
+        value = await cache.aget(test_key)
+        await cache.adelete(test_key)
+        
+        latency = (time.time() - start) * 1000
+        
+        result["status"] = "healthy" if value else "degraded"
+        result["latency_ms"] = round(latency, 2)
+        
+    except Exception as e:
+        result["status"] = "unhealthy"
+        result["error"] = str(e)
+    
+    return result

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
   "lightning>=2.6.0",
   "fiddle>=0.3.0",
   "cloudpickle>=3.1.2",
+  "upstash-redis>=1.5.0",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -1,15 +1,23 @@
 # This file was autogenerated by uv via the following command:
 #    uv pip compile pyproject.toml -o requirements.txt --python-version 3.12
+absl-py==2.3.1
+    # via fiddle
 aiohappyeyeballs==2.6.1
     # via aiohttp
 aiohttp==3.13.2
-    # via langchain-nvidia-ai-endpoints
+    # via
+    #   fsspec
+    #   langchain-nvidia-ai-endpoints
 aiosignal==1.4.0
     # via aiohttp
 annotated-doc==0.0.4
     # via fastapi
 annotated-types==0.7.0
     # via pydantic
+antlr4-python3-runtime==4.9.3
+    # via
+    #   hydra-core
+    #   omegaconf
 anyio==4.12.0
     # via
     #   httpx
@@ -24,6 +32,7 @@ certifi==2025.11.12
     # via
     #   httpcore
     #   httpx
+    #   pinecone
     #   requests
 cffi==2.0.0
     # via cryptography
@@ -31,17 +40,24 @@ charset-normalizer==3.4.4
     # via requests
 click==8.3.1
     # via uvicorn
+cloudpickle==3.1.2
+    # via ideasprinter-api (pyproject.toml)
 colorama==0.4.6
-    # via click
+    # via
+    #   click
+    #   tqdm
 cryptography==46.0.3
     # via python-jose
 defusedxml==0.7.1
     # via fpdf2
 ecdsa==0.19.1
     # via python-jose
-    # via ideasprinter-api (pyproject.toml)
 fastapi==0.123.5
     # via ideasprinter-api (pyproject.toml)
+fiddle==0.3.0
+    # via ideasprinter-api (pyproject.toml)
+filelock==3.20.2
+    # via torch
 filetype==1.2.0
     # via langchain-nvidia-ai-endpoints
 fonttools==4.61.0
@@ -52,6 +68,11 @@ frozenlist==1.8.0
     # via
     #   aiohttp
     #   aiosignal
+fsspec==2025.12.0
+    # via
+    #   lightning
+    #   pytorch-lightning
+    #   torch
 google-api-core==2.28.1
     # via google-api-python-client
 google-api-python-client==2.187.0
@@ -68,6 +89,8 @@ google-auth-oauthlib==1.2.3
     # via ideasprinter-api (pyproject.toml)
 googleapis-common-protos==1.72.0
     # via google-api-core
+graphviz==0.21
+    # via fiddle
 greenlet==3.3.0
     # via sqlalchemy
 h11==0.16.0
@@ -84,6 +107,9 @@ httpx==0.28.1
     # via
     #   langgraph-sdk
     #   langsmith
+    #   upstash-redis
+hydra-core==1.3.2
+    # via ideasprinter-api (pyproject.toml)
 idna==3.11
     # via
     #   anyio
@@ -91,7 +117,9 @@ idna==3.11
     #   requests
     #   yarl
 jinja2==3.1.6
-    # via ideasprinter-api (pyproject.toml)
+    # via
+    #   ideasprinter-api (pyproject.toml)
+    #   torch
 jsonpatch==1.33
     # via langchain-core
 jsonpointer==3.0.0
@@ -122,31 +150,60 @@ langsmith==0.4.53
     # via
     #   ideasprinter-api (pyproject.toml)
     #   langchain-core
+libcst==1.8.6
+    # via fiddle
+lightning==2.6.0
+    # via ideasprinter-api (pyproject.toml)
+lightning-utilities==0.15.2
+    # via
+    #   lightning
+    #   pytorch-lightning
+    #   torchmetrics
 markupsafe==3.0.3
     # via jinja2
+mpmath==1.3.0
+    # via sympy
 multidict==6.7.0
     # via
     #   aiohttp
     #   yarl
+networkx==3.6.1
+    # via torch
 numpy==2.3.5
     # via
     #   ideasprinter-api (pyproject.toml)
+    #   torchmetrics
 oauthlib==3.3.1
     # via requests-oauthlib
+omegaconf==2.3.0
+    # via hydra-core
 orjson==3.11.4
     # via
     #   langgraph-sdk
     #   langsmith
+    #   pinecone
 ormsgpack==1.12.0
     # via langgraph-checkpoint
-packaging==25.0
+packaging==24.2
     # via
+    #   hydra-core
     #   langchain-core
     #   langsmith
+    #   lightning
+    #   lightning-utilities
+    #   pinecone-plugin-assistant
+    #   pytorch-lightning
+    #   torchmetrics
 passlib==1.7.4
     # via ideasprinter-api (pyproject.toml)
 pillow==12.0.0
     # via fpdf2
+pinecone==8.0.0
+    # via ideasprinter-api (pyproject.toml)
+pinecone-plugin-assistant==3.0.1
+    # via pinecone
+pinecone-plugin-interface==0.0.7
+    # via pinecone
 propcache==0.4.1
     # via
     #   aiohttp
@@ -180,18 +237,28 @@ pydantic-core==2.41.5
     # via pydantic
 pyparsing==3.2.5
     # via httplib2
+python-dateutil==2.9.0.post0
+    # via pinecone
 python-dotenv==1.2.1
     # via ideasprinter-api (pyproject.toml)
 python-jose==3.5.0
     # via ideasprinter-api (pyproject.toml)
 python-multipart==0.0.20
     # via ideasprinter-api (pyproject.toml)
+pytorch-lightning==2.6.0
+    # via lightning
 pyyaml==6.0.3
-    # via langchain-core
+    # via
+    #   langchain-core
+    #   libcst
+    #   lightning
+    #   omegaconf
+    #   pytorch-lightning
 requests==2.32.5
     # via
     #   google-api-core
     #   langsmith
+    #   pinecone-plugin-assistant
     #   requests-oauthlib
     #   requests-toolbelt
 requests-oauthlib==2.0.0
@@ -202,31 +269,62 @@ rsa==4.9.1
     # via
     #   google-auth
     #   python-jose
+setuptools==80.9.0
+    # via
+    #   lightning-utilities
+    #   torch
 six==1.17.0
-    # via ecdsa
+    # via
+    #   ecdsa
+    #   python-dateutil
 sqlalchemy==2.0.45
     # via ideasprinter-api (pyproject.toml)
 starlette==0.50.0
     # via fastapi
+sympy==1.14.0
+    # via torch
 tenacity==9.1.2
     # via langchain-core
+torch==2.9.1
+    # via
+    #   lightning
+    #   pytorch-lightning
+    #   torchmetrics
+torchmetrics==1.8.2
+    # via
+    #   lightning
+    #   pytorch-lightning
+tqdm==4.67.1
+    # via
+    #   lightning
+    #   pytorch-lightning
 typing-extensions==4.15.0
     # via
     #   aiosignal
     #   anyio
     #   fastapi
+    #   fiddle
     #   langchain-core
+    #   lightning
+    #   lightning-utilities
+    #   pinecone
     #   pydantic
     #   pydantic-core
+    #   pytorch-lightning
     #   sqlalchemy
     #   starlette
+    #   torch
     #   typing-inspection
 typing-inspection==0.4.2
     # via pydantic
+upstash-redis==1.5.0
+    # via ideasprinter-api (pyproject.toml)
 uritemplate==4.2.0
     # via google-api-python-client
 urllib3==2.5.0
-    # via requests
+    # via
+    #   pinecone
+    #   requests
 uuid-utils==0.12.0
     # via langsmith
 uvicorn==0.38.0

uv.lock

@@ -903,6 +903,7 @@ dependencies = [
     { name = "python-jose", extra = ["cryptography"] },
     { name = "python-multipart" },
     { name = "sqlalchemy" },
+    { name = "upstash-redis" },
     { name = "uvicorn" },
 ]
 
@@ -941,6 +942,7 @@ requires-dist = [
     { name = "python-jose", extras = ["cryptography"], specifier = ">=3.3.0" },
     { name = "python-multipart", specifier = ">=0.0.6" },
     { name = "sqlalchemy", specifier = ">=2.0.0" },
+    { name = "upstash-redis", specifier = ">=1.5.0" },
     { name = "uvicorn", specifier = ">=0.30.0" },
 ]
 provides-extras = ["dev"]
@@ -2535,6 +2537,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/dc/9b/47798a6c91d8bdb567fe2698fe81e0c6b7cb7ef4d13da4114b41d239f65d/typing_inspection-0.4.2-py3-none-any.whl", hash = "sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7", size = 14611, upload-time = "2025-10-01T02:14:40.154Z" },
 ]
 
+[[package]]
+name = "upstash-redis"
+version = "1.5.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "httpx" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/a8/62/bc53c35fbf4e2b774ab0eb02f3908cfe89b6636e87cdc40b264a4fc1dcce/upstash_redis-1.5.0.tar.gz", hash = "sha256:1917d4d009ca803815092892d92c7da9138b4ada6b353974fb74caf063c6d2a3", size = 39356, upload-time = "2025-10-22T10:15:34.608Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5a/87/d24541a1d9c29033e74aa05b5d8b4857feff79344ebd8fca410eb4683795/upstash_redis-1.5.0-py3-none-any.whl", hash = "sha256:e08de1f74d3fb48a81b383c00398cc9336c43b65b82e6d9266312143970800b9", size = 41088, upload-time = "2025-10-22T10:15:33.363Z" },
+]
+
 [[package]]
 name = "uritemplate"
 version = "4.2.0"