Add multi-year budget-window API v1 support

changelog.d/budget-window-batch.fixed.md

@@ -0,0 +1 @@
+Added a budget-window economy endpoint that batches yearly impact calculations with bounded server-side concurrency and returns aggregated progress plus totals.

changelog.d/fix-silent-exception-swallowing.fixed.md

@@ -0,0 +1 @@
+Log exceptions instead of silently swallowing them during household calculations.

policyengine_api/api.py

@@ -4,6 +4,7 @@
 
 import time
 import sys
+import os
 
 start_time = time.time()
 
@@ -157,8 +158,11 @@ def log_timing(message):
 app.register_blueprint(user_profile_bp)
 log_timing("User profile routes registered")
 
-app.route("/simulations", methods=["GET"])(get_simulations)
-log_timing("Simulations endpoint registered")
+if os.environ.get("FLASK_DEBUG") == "1":
+    app.route("/simulations", methods=["GET"])(get_simulations)
+    log_timing("Simulations endpoint registered")
+else:
+    log_timing("Simulations endpoint skipped outside debug mode")
 
 app.register_blueprint(tracer_analysis_bp)
 log_timing("Tracer analysis routes registered")

policyengine_api/country.py

@@ -1,5 +1,6 @@
 import importlib
 import inspect
+import logging
 import json
 from policyengine_core.taxbenefitsystems import TaxBenefitSystem
 from typing import Union, Optional
@@ -429,11 +430,9 @@ def calculate(
                         entity_result
                     )
             except Exception as e:
-                if "axes" in household:
-                    pass
-                else:
+                logging.exception(f"Error computing {variable_name} for {entity_id}")
+                if "axes" not in household:
                     household[entity_plural][entity_id][variable_name][period] = None
-                    print(f"Error computing {variable_name} for {entity_id}: {e}")
 
         tracer_output = simulation.tracer.computation_log
         log_lines = tracer_output.lines(aggregate=False, max_depth=10)

policyengine_api/data/__init__.py

@@ -1 +1,6 @@
-from .data import PolicyEngineDatabase, database, local_database
+from .data import (
+    PolicyEngineDatabase,
+    database,
+    get_remote_database,
+    local_database,
+)

policyengine_api/data/data.py

@@ -19,6 +19,7 @@ class _ResultProxy:
     Provides fetchone()/fetchall() with dict-like row access."""
 
     def __init__(self, cursor_result):
+        self.rowcount = getattr(cursor_result, "rowcount", -1)
         try:
             # Use .mappings() so rows behave like dicts
             self._rows = list(cursor_result.mappings())
@@ -105,16 +106,20 @@ def _create_pool(self):
             with open(".dbpw") as f:
                 db_pass = f.read().strip()
         db_name = "policyengine"
-        conn = self.connector.connect(
-            instance_connection_string=instance_connection_name,
-            driver="pymysql",
-            db=db_name,
-            user=db_user,
-            password=db_pass,
-        )
+
+        def get_connection():
+            return self.connector.connect(
+                instance_connection_string=instance_connection_name,
+                driver="pymysql",
+                db=db_name,
+                user=db_user,
+                password=db_pass,
+            )
+
         self.pool = sqlalchemy.create_engine(
             "mysql+pymysql://",
-            creator=lambda: conn,
+            creator=get_connection,
+            pool_pre_ping=True,
         )
 
     def _close_pool(self):
@@ -259,3 +264,11 @@ def initialize(self):
     database = PolicyEngineDatabase(local=False, initialize=False)
 
 local_database = PolicyEngineDatabase(local=True, initialize=False)
+remote_database = None
+
+
+def get_remote_database() -> PolicyEngineDatabase:
+    global remote_database
+    if remote_database is None:
+        remote_database = PolicyEngineDatabase(local=False, initialize=False)
+    return remote_database

policyengine_api/endpoints/simulation.py

@@ -1,4 +1,4 @@
-from policyengine_api.data import local_database
+from policyengine_api.data import get_remote_database
 
 """
 
@@ -42,10 +42,14 @@ def get_simulations(
         max_results = _DEFAULT_SIMULATION_RESULTS
     max_results = max(1, min(max_results, _MAX_SIMULATION_RESULTS))
 
-    result = local_database.query(
-        "SELECT * FROM reform_impact ORDER BY start_time DESC LIMIT ?",
-        (max_results,),
-    ).fetchall()
+    result = (
+        get_remote_database()
+        .query(
+            "SELECT * FROM reform_impact ORDER BY start_time DESC LIMIT ?",
+            (max_results,),
+        )
+        .fetchall()
+    )
 
     # Format into [{}]
 

policyengine_api/libs/simulation_api_modal.py

@@ -7,7 +7,7 @@
 
 import os
 import sys
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Optional
 
 import httpx
@@ -42,6 +42,28 @@ def name(self) -> str:
         return self.job_id
 
 
+@dataclass
+class ModalBudgetWindowBatchExecution:
+    """
+    Represents a budget-window batch execution in the Modal simulation API.
+    """
+
+    batch_job_id: str
+    status: str
+    progress: Optional[int] = None
+    completed_years: list[str] = field(default_factory=list)
+    running_years: list[str] = field(default_factory=list)
+    queued_years: list[str] = field(default_factory=list)
+    failed_years: list[str] = field(default_factory=list)
+    result: Optional[dict] = None
+    error: Optional[str] = None
+
+    @property
+    def name(self) -> str:
+        """Alias for batch_job_id."""
+        return self.batch_job_id
+
+
 class SimulationAPIModal:
     """
     HTTP client for the Modal Simulation API.
@@ -144,10 +166,51 @@ def run(self, payload: dict) -> ModalSimulationExecution:
             )
             raise
 
+    def run_budget_window_batch(self, payload: dict) -> ModalBudgetWindowBatchExecution:
+        """
+        Submit a budget-window batch job to the Modal API.
+        """
+        try:
+            modal_payload = dict(payload)
+            if "model_version" in modal_payload:
+                modal_payload["version"] = modal_payload.pop("model_version")
+            modal_payload.pop("data_version", None)
+
+            response = self.client.post(
+                f"{self.base_url}/simulate/economy/budget-window",
+                json=modal_payload,
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            logger.log_struct(
+                {
+                    "message": "Modal budget-window batch submitted",
+                    "batch_job_id": data.get("batch_job_id"),
+                    "status": data.get("status"),
+                },
+                severity="INFO",
+            )
+
+            return ModalBudgetWindowBatchExecution(
+                batch_job_id=data["batch_job_id"],
+                status=data["status"],
+            )
+
+        except httpx.HTTPStatusError as e:
+            logger.log_struct(
+                {
+                    "message": f"Modal batch API HTTP error: {e.response.status_code}",
+                    "response_text": e.response.text[:500],
+                },
+                severity="ERROR",
+            )
+            raise
+
         except httpx.RequestError as e:
             logger.log_struct(
                 {
-                    "message": f"Modal API request error: {str(e)}",
+                    "message": f"Modal batch API request error: {str(e)}",
                     "run_id": (payload.get("_telemetry") or {}).get("run_id"),
                 },
                 severity="ERROR",
@@ -226,10 +289,44 @@ def get_execution_by_id(self, job_id: str) -> ModalSimulationExecution:
             )
             raise
 
+    def get_budget_window_batch_by_id(
+        self, batch_job_id: str
+    ) -> ModalBudgetWindowBatchExecution:
+        """
+        Poll the Modal API for the current status of a budget-window batch.
+        """
+        try:
+            response = self.client.get(
+                f"{self.base_url}/budget-window-jobs/{batch_job_id}"
+            )
+            data = response.json()
+
+            return ModalBudgetWindowBatchExecution(
+                batch_job_id=batch_job_id,
+                status=data["status"],
+                progress=data.get("progress"),
+                completed_years=data.get("completed_years", []),
+                running_years=data.get("running_years", []),
+                queued_years=data.get("queued_years", []),
+                failed_years=data.get("failed_years", []),
+                result=data.get("result"),
+                error=data.get("error"),
+            )
+
+        except httpx.HTTPStatusError as e:
+            logger.log_struct(
+                {
+                    "message": f"Modal batch API HTTP error polling job {batch_job_id}: {e.response.status_code}",
+                    "response_text": e.response.text[:500],
+                },
+                severity="ERROR",
+            )
+            raise
+
         except httpx.RequestError as e:
             logger.log_struct(
                 {
-                    "message": f"Modal API request error polling job {job_id}: {str(e)}",
+                    "message": f"Modal batch API request error polling job {batch_job_id}: {str(e)}",
                 },
                 severity="ERROR",
             )