junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Pace Solar calls per container to stay under the 600 QPM fleet cap 🟩

Browse source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-25 16:00:33 +00:00
parent 48a47590e9
commit 4b9bd495e1
1 changed files with 19 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
infrastructure/solar/google_solar_api_client.py
View file

@ -39,6 +39,24 @@ class GoogleSolarApiClient:
self._min_request_interval_seconds = min_request_interval_seconds self._min_request_interval_seconds = min_request_interval_seconds
self._last_request_monotonic: Optional[float] = None self._last_request_monotonic: Optional[float] = None
def _await_rate_limit(self) -> None:
"""Block until at least ``min_request_interval_seconds`` has elapsed since
this instance's previous call, so the container never exceeds its share of
the fleet's 600 QPM Solar budget. A no-op when the interval is 0 (the
default / interactive path). The call's own latency counts toward the gap
(we sleep only the *remainder*), so we don't double-pay the request time."""
if self._min_request_interval_seconds <= 0:
return
now = time.monotonic()
if self._last_request_monotonic is not None:
wait = self._min_request_interval_seconds - (
now - self._last_request_monotonic
)
if wait > 0:
time.sleep(wait)
now = time.monotonic()
self._last_request_monotonic = now
@staticmethod @staticmethod
def _parse_retry_after(response: requests.Response) -> Optional[float]: def _parse_retry_after(response: requests.Response) -> Optional[float]:
header = response.headers.get("Retry-After") header = response.headers.get("Retry-After")
@ -55,6 +73,7 @@ class GoogleSolarApiClient:
latitude: float, latitude: float,
required_quality: Literal["HIGH", "MEDIUM", "LOW"] = "MEDIUM", required_quality: Literal["HIGH", "MEDIUM", "LOW"] = "MEDIUM",
) -> dict[str, Any]: ) -> dict[str, Any]:
self._await_rate_limit()
insights_url = f"{self.base_url}/buildingInsights:findClosest" insights_url = f"{self.base_url}/buildingInsights:findClosest"
params: dict[str, str] = { params: dict[str, str] = {
"location.latitude": f"{latitude:.5f}", "location.latitude": f"{latitude:.5f}",