Fix cache_table to read from cache before API call in estimates/flows

The cache_table parameter was write-only — it stored results but never
checked for cached data before making API calls. Also adds variables,
breakdown, and year to cache keys to prevent collisions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: chekos

pypums/estimates.py

@@ -131,6 +131,21 @@ def get_estimates(
     if year is not None:
         params["YEAR"] = str(year)
 
+    # Build cache key from all parameters that affect the result.
+    vars_str = ",".join(variables) if variables is not None else ""
+    breakdown_str = ",".join(breakdown) if breakdown is not None else ""
+    cache_key = (
+        f"est_{vintage}_{resolved_product}_{geography}_{state}_{county}"
+        f"_{year}_{vars_str}_{breakdown_str}"
+    )
+
+    # Check cache before calling API.
+    if cache_table:
+        disk_cache = CensusCache(_DEFAULT_CACHE_DIR)
+        cached = disk_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
     if show_call:
         print(f"Census API call: {url}")
         print(f"  Parameters: {params}")
@@ -158,7 +173,6 @@ def get_estimates(
         df = attach_geometry(df, geography, state=state, year=vintage)
 
     if cache_table:
-        cache_key = f"est_{vintage}_{resolved_product}_{geography}_{state}_{county}"
         disk_cache = CensusCache(_DEFAULT_CACHE_DIR)
         disk_cache.set(cache_key, df, ttl_seconds=86400)
 

pypums/flows.py

@@ -141,6 +141,21 @@ def get_flows(
         for dim in breakdown:
             params[dim] = "*"
 
+    # Build cache key from all parameters that affect the result.
+    vars_str = ",".join(variables) if variables is not None else ""
+    breakdown_str = ",".join(breakdown) if breakdown is not None else ""
+    cache_key = (
+        f"flows_{year}_{geography}_{state}_{county}_{msa}"
+        f"_{vars_str}_{breakdown_str}"
+    )
+
+    # Check cache before calling API.
+    if cache_table:
+        disk_cache = CensusCache(_DEFAULT_CACHE_DIR)
+        cached = disk_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
     if show_call:
         print(f"Census API call: {url}")
         print(f"  Parameters: {params}")
@@ -157,7 +172,6 @@ def get_flows(
             df[col] = pd.to_numeric(df[col], errors="coerce")
 
     if cache_table:
-        cache_key = f"flows_{year}_{geography}_{state}_{county}_{msa}"
         disk_cache = CensusCache(_DEFAULT_CACHE_DIR)
         disk_cache.set(cache_key, df, ttl_seconds=86400)