⚡ perf: Add Redis caching to /v1/check-email with privacy-safe email hashing

Author: DevaOnBreaches

api/v1/breaches.py

@@ -1,6 +1,7 @@
 """Breach-related API endpoints."""
 
 # Standard library imports
+import hashlib
 import json
 from datetime import datetime, timedelta
 from typing import Dict, Optional, Union
@@ -81,6 +82,11 @@ def cache_breaches(
         pass
 
 
+def hash_email(email: str) -> str:
+    """Hash email for privacy-safe cache keys."""
+    return hashlib.sha256(email.lower().encode()).hexdigest()[:16]
+
+
 @router.get("/breaches", response_model=BreachListResponse)
 @custom_rate_limiter("2 per second;50 per hour;100 per day")
 async def get_xposed_breaches(
@@ -436,11 +442,8 @@ async def search_email(
             return EmailBreachErrorResponse(Error="Not found")
 
         email = email.lower()
-        breach_data = await get_exposure(email)
-
-        if not breach_data:
-            return EmailBreachErrorResponse(Error="Not found")
 
+        # Always check shieldOn first (privacy - can't cache this)
         data_store = datastore.Client()
         alert_key = data_store.key("xon_alert", email)
         alert_record = data_store.get(alert_key)
@@ -455,6 +458,19 @@ async def search_email(
                 },
             )
 
+        # Check cache (after shieldOn check passes)
+        cache_key = f"check-email:{hash_email(email)}:{details}"
+        cached_result = get_cached_breaches(cache_key)
+        if cached_result:
+            # Replace hashed email with actual email in response
+            cached_result["email"] = email
+            return JSONResponse(status_code=200, content=cached_result)
+
+        breach_data = await get_exposure(email)
+
+        if not breach_data:
+            return EmailBreachErrorResponse(Error="Not found")
+
         xon_key = data_store.key("xon", email)
         xon_record = data_store.get(xon_key)
 
@@ -503,6 +519,11 @@ async def search_email(
 
                     response_content["breach_details"] = formatted_breaches
 
+                # Cache the response (use placeholder for email to avoid storing PII)
+                cache_content = response_content.copy()
+                cache_content["email"] = "__cached__"
+                cache_breaches(cache_key, cache_content)
+
                 return JSONResponse(status_code=200, content=response_content)
 
         return JSONResponse(

api/v1/metrics.py

@@ -180,6 +180,13 @@ async def get_domain_metrics(request: Request, domain: str) -> JSONResponse:
         if not validate_url(request):
             raise HTTPException(status_code=400, detail="Invalid request URL")
 
+        # Check cache first
+        cache_key = f"metrics:domain:{domain.lower()}"
+        cached_result = get_cached_metrics(cache_key)
+        if cached_result:
+            return JSONResponse(content=cached_result)
+
+        # Cache miss - build response
         domain_metrics = {
             "status": "success",
             "message": "Domain metrics retrieved successfully",
@@ -195,6 +202,8 @@ async def get_domain_metrics(request: Request, domain: str) -> JSONResponse:
             },
         }
 
+        cache_metrics(cache_key, domain_metrics)
+
         return JSONResponse(content=domain_metrics)
 
     except Exception as e: