perf: parallel BQ queries + TTL cache on clusters/quality/analytics

Author: StephaneWamba

api/services/analytics_service.py

@@ -1,67 +1,77 @@
-"""Analytics data — queries the 6 looker_* BigQuery views."""
+"""Analytics data — 6 BigQuery views queried in parallel with TTL cache."""
 from __future__ import annotations
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from google.cloud import bigquery
 from config import PROJECT, DATASET
+from services.cache import ttl_cache
 
+_ds = f"{PROJECT}.{DATASET}"
 
-def get_analytics(bq: bigquery.Client) -> dict:
-    ds = f"{PROJECT}.{DATASET}"
-
-    cluster_dist = list(bq.query(f"""
+_QUERIES = {
+    "cluster_distribution": f"""
         SELECT cluster_label, product_count, avg_price
-        FROM `{ds}.looker_cluster_distribution`
-        ORDER BY product_count DESC
-        LIMIT 40
-    """).result())
-
-    pricing = list(bq.query(f"""
+        FROM `{_ds}.looker_cluster_distribution`
+        ORDER BY product_count DESC LIMIT 40""",
+    "pricing": f"""
         SELECT cluster_label, avg_price, price_min, price_max, product_count
-        FROM `{ds}.looker_pricing_per_cluster`
-        ORDER BY avg_price DESC
-        LIMIT 20
-    """).result())
-
-    heatmap = list(bq.query(f"""
+        FROM `{_ds}.looker_pricing_per_cluster`
+        ORDER BY avg_price DESC LIMIT 20""",
+    "heatmap": f"""
         SELECT category, department, product_count, avg_price
-        FROM `{ds}.looker_heatmap_cat_dept`
-        ORDER BY product_count DESC
-    """).result())
-
-    quality = list(bq.query(f"""
+        FROM `{_ds}.looker_heatmap_cat_dept`
+        ORDER BY product_count DESC""",
+    "quality": f"""
         SELECT completeness_pct, total_records, valid_records,
                field_name_completeness, field_brand_completeness,
                field_cat_completeness, field_price_completeness,
                price_mean, price_min, price_max
-        FROM `{ds}.looker_data_quality`
-        LIMIT 1
-    """).result())
-
-    timeline = list(bq.query(f"""
+        FROM `{_ds}.looker_data_quality` LIMIT 1""",
+    "timeline": f"""
         SELECT sale_date, cluster_label, sales_count, sales_revenue
-        FROM `{ds}.looker_sales_timeline`
-        ORDER BY sale_date ASC
-    """).result())
-
-    brands = list(bq.query(f"""
+        FROM `{_ds}.looker_sales_timeline`
+        ORDER BY sale_date ASC""",
+    "brands": f"""
         SELECT cluster_label, brand, product_count
-        FROM `{ds}.looker_brands_per_cluster`
-        ORDER BY cluster_label, product_count DESC
-        LIMIT 100
-    """).result())
+        FROM `{_ds}.looker_brands_per_cluster`
+        ORDER BY cluster_label, product_count DESC LIMIT 100""",
+}
+
+
+def _row(r) -> dict:
+    return {
+        k: (float(v) if hasattr(v, "__float__") and not isinstance(v, (int, str, bool)) else v)
+        for k, v in dict(r).items()
+    }
+
+
+def _run(bq: bigquery.Client, key: str) -> tuple[str, list]:
+    rows = list(bq.query(_QUERIES[key]).result())
+    return key, rows
+
+
+@ttl_cache(seconds=600)  # 10-minute cache — data only changes after pipeline run
+def get_analytics(bq: bigquery.Client) -> dict:
+    results: dict[str, list] = {}
+
+    # Fire all 6 queries in parallel
+    with ThreadPoolExecutor(max_workers=6) as pool:
+        futures = {pool.submit(_run, bq, key): key for key in _QUERIES}
+        for future in as_completed(futures):
+            key, rows = future.result()
+            results[key] = rows
 
-    def row(r):
-        return {k: (float(v) if hasattr(v, '__float__') and not isinstance(v, (int, str, bool)) else v)
-                for k, v in dict(r).items()}
+    timeline_rows = results["timeline"]
+    quality_rows = results["quality"]
 
     return {
-        "cluster_distribution": [row(r) for r in cluster_dist],
-        "pricing": [row(r) for r in pricing],
-        "heatmap": [row(r) for r in heatmap],
-        "quality": row(quality[0]) if quality else {},
+        "cluster_distribution": [_row(r) for r in results["cluster_distribution"]],
+        "pricing": [_row(r) for r in results["pricing"]],
+        "heatmap": [_row(r) for r in results["heatmap"]],
+        "quality": _row(quality_rows[0]) if quality_rows else {},
         "timeline": [
             {**{k: v for k, v in dict(r).items() if k != "sale_date"},
              "sale_date": str(r.sale_date)}
-            for r in timeline
+            for r in timeline_rows
         ],
-        "brands": [row(r) for r in brands],
+        "brands": [_row(r) for r in results["brands"]],
     }

api/services/cache.py

@@ -0,0 +1,23 @@
+"""Simple in-memory TTL cache for expensive BigQuery results."""
+from __future__ import annotations
+import time
+import functools
+from typing import Any
+
+_store: dict[str, tuple[float, Any]] = {}
+
+
+def ttl_cache(seconds: int = 600):
+    """Decorator: cache function result for `seconds`. Key = func name + args."""
+    def decorator(fn):
+        @functools.wraps(fn)
+        def wrapper(*args, **kwargs):
+            key = f"{fn.__qualname__}:{args[1:]}:{kwargs}"
+            entry = _store.get(key)
+            if entry and time.monotonic() - entry[0] < seconds:
+                return entry[1]
+            result = fn(*args, **kwargs)
+            _store[key] = (time.monotonic(), result)
+            return result
+        return wrapper
+    return decorator

api/services/cluster_service.py

@@ -4,8 +4,10 @@
 from google.cloud import bigquery
 from config import TABLE_CLUSTERED, TABLE_CLEAN, TABLE_ENRICHED
 from models.schemas import ClusterSummary, ClustersResponse, ClusterProduct, ClusterProductsResponse
+from services.cache import ttl_cache
 
 
+@ttl_cache(seconds=600)
 def get_clusters(bq: bigquery.Client) -> ClustersResponse:
     sql = f"""
     SELECT
@@ -32,6 +34,7 @@ def get_clusters(bq: bigquery.Client) -> ClustersResponse:
     return ClustersResponse(clusters=clusters)
 
 
+@ttl_cache(seconds=600)
 def get_cluster_products(bq: bigquery.Client, cluster_id: int) -> ClusterProductsResponse:
     sql = f"""
     SELECT

api/services/quality_service.py

@@ -4,8 +4,10 @@
 from google.cloud import bigquery
 from config import TABLE_QUALITY
 from models.schemas import QualityReport
+from services.cache import ttl_cache
 
 
+@ttl_cache(seconds=600)
 def get_quality_report(bq: bigquery.Client) -> QualityReport:
     sql = f"""
     SELECT *