From 5142122548ffae072381771af6a08e93adb391e2 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Tue, 1 Sep 2026 14:53:03 -0400 Subject: [PATCH 1/5] Fix include_imputed_marker being passed as include_meta in get_historical_csv get_historical_csv passed include_imputed_marker as the sixth positional argument to get_historical_pandas, whose sixth parameter is include_meta. As a result, get_historical_csv(..., include_imputed_marker=True) wrote a CSV containing a meta column and no imputed_data_used column. Pass all defaulted parameters by keyword in the internal call chain (get_historical_csv -> get_historical_pandas -> get_historical_jsons) so argument order differences between signatures cannot misroute a flag. Public signatures are unchanged. Adds a regression test asserting the CSV contains imputed_data_used and not meta when include_imputed_marker=True. Co-Authored-By: Claude Fable 5 --- tests/test_sdk.py | 18 ++++++++++++++++++ watttime/api.py | 14 ++++++++++++-- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 557cee3a..a79ab392 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -289,6 +289,24 @@ def test_get_historical_csv(self): assert fp.exists() fp.unlink() + def test_get_historical_csv_include_imputed(self): + start = parse("2025-01-01 00:00Z") + end = parse("2025-01-02 00:00Z") + self.historical.get_historical_csv( + start, end, REGION, include_imputed_marker=True + ) + + fp = ( + Path.home() + / "watttime_historical_csvs" + / f"{REGION}_co2_moer_{start.date()}_{end.date()}.csv" + ) + assert fp.exists() + df = pd.read_csv(fp) + self.assertIn("imputed_data_used", df.columns) + self.assertNotIn("meta", df.columns) + fp.unlink() + def test_multi_model_range(self): """If model is not specified, we should only return the most recent model data""" myaccess = WattTimeMyAccess() diff --git a/watttime/api.py b/watttime/api.py index 81d372c4..7f04fae5 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -467,7 +467,12 @@ def get_historical_pandas( pd.DataFrame: _description_ """ responses = self.get_historical_jsons( - start, end, region, signal_type, model, include_imputed_marker + start, + end, + region, + signal_type=signal_type, + model=model, + include_imputed_marker=include_imputed_marker, ) df = pd.json_normalize( responses, record_path="data", meta=["meta"] if include_meta else [] @@ -503,7 +508,12 @@ def get_historical_csv( None, results are saved to a csv file in the user's home directory. """ df = self.get_historical_pandas( - start, end, region, signal_type, model, include_imputed_marker + start, + end, + region, + signal_type=signal_type, + model=model, + include_imputed_marker=include_imputed_marker, ) out_dir = Path.home() / "watttime_historical_csvs" From 9d334f8a7ed718b57f66151dbcb02a43c801d88d Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:15:44 -0400 Subject: [PATCH 2/5] Expose request timeout and historical chunk size; hint at both on timeout Add a keyword-only `timeout` to WattTimeBase.__init__ (default (10, 60), matching the existing login/register literals) and pass it at all three HTTP call sites. The data request previously used a scalar 60, so its default connect timeout drops from 60 s to 10 s; default read timeouts are unchanged. Add a keyword-only `chunk_size` to get_historical_jsons/_pandas/_csv, forwarded to _get_chunks, which now accepts None (30 days) and rejects sizes of 5 minutes or less -- the per-chunk trim -- instead of looping forever or producing inverted chunks. When a request fails on a timeout, append a hint to the RuntimeError naming both knobs. Once the session's retries are exhausted, requests raises ConnectionError -> MaxRetryError -> ReadTimeoutError rather than ReadTimeout, so _is_timeout walks the exception chain. Motivation: in August 2026, 30-day historical pulls for some regions took over 180 s server-side while the same pulls at 10 days took ~11 s. Users hit an opaque read timeout after ~4 minutes with no supported lever; the only workaround was monkeypatching _get_chunks. Co-Authored-By: Claude Fable 5.1 --- tests/test_sdk.py | 176 ++++++++++++++++++++++++++++++++++++++++++++++ watttime/api.py | 84 +++++++++++++++++++--- 2 files changed, 249 insertions(+), 11 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index a79ab392..9db18a42 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -1,3 +1,4 @@ +import inspect import unittest import unittest.mock as mock from unittest.mock import patch @@ -15,6 +16,8 @@ ) from pathlib import Path import pytest +import requests +import urllib3.exceptions from shapely.geometry import shape, Polygon, MultiPolygon import pandas as pd @@ -74,6 +77,119 @@ def test_login_sets_user_agent_header(self): self.base.headers["User-Agent"], f"watttime-python-sdk-{VERSION}" ) + def _mock_login_response(self): + mock_rsp = mock.Mock() + mock_rsp.json.return_value = {"token": "fake-token"} + mock_rsp.raise_for_status.return_value = None + return mock_rsp + + def _authed_base(self, **kwargs): + """A client with a fake, unexpired token so requests skip _login().""" + base = WattTimeBase(**kwargs) + base.token = "fake-token" + base.token_valid_until = datetime.now() + timedelta(minutes=30) + base.headers = {"Authorization": "Bearer fake-token"} + return base + + def test_timeout_reaches_all_three_call_sites(self): + """The default (10, 60) is passed to login, the data request, and register.""" + base = WattTimeBase() + self.assertEqual(base.timeout, (10, 60)) + + mock_rsp = mock.Mock() + mock_rsp.json.return_value = {"token": "fake-token", "data": [], "meta": {}} + mock_rsp.raise_for_status.return_value = None + + with mock.patch.object(base.session, "get", return_value=mock_rsp) as mock_get: + base._login() + self.assertEqual(mock_get.call_args.kwargs["timeout"], (10, 60)) + + base._make_rate_limited_request("https://api.watttime.org/v3/test", {}) + self.assertEqual(mock_get.call_args.kwargs["timeout"], (10, 60)) + + with mock.patch.object( + base.session, "post", return_value=mock_rsp + ) as mock_post: + base.register(email="someone@example.com") + self.assertEqual(mock_post.call_args.kwargs["timeout"], (10, 60)) + + base.session.close() + + def test_timeout_accepts_tuple_and_scalar_and_none(self): + """Both `requests` timeout forms, and None, are stored and forwarded as given.""" + for value in [(3, 5), 30, 12.5, None]: + with self.subTest(timeout=value): + base = WattTimeBase(timeout=value) + self.assertEqual(base.timeout, value) + + with mock.patch.object( + base.session, "get", return_value=self._mock_login_response() + ) as mock_get: + base._login() + + self.assertEqual(mock_get.call_args.kwargs["timeout"], value) + base.session.close() + + def test_new_parameters_are_keyword_only(self): + """New optional parameters are keyword-only, so existing positional calls + are unaffected and the positional contract stays where it is today.""" + keyword_only = inspect.Parameter.KEYWORD_ONLY + checks = [ + (WattTimeBase.__init__, "timeout"), + (WattTimeHistorical.get_historical_jsons, "chunk_size"), + (WattTimeHistorical.get_historical_pandas, "chunk_size"), + (WattTimeHistorical.get_historical_csv, "chunk_size"), + ] + for func, name in checks: + with self.subTest(f"{func.__qualname__}({name})"): + param = inspect.signature(func).parameters[name] + self.assertIs(param.kind, keyword_only) + + client = WattTimeHistorical(None, None, False, 10, 4) + self.assertEqual(client.worker_count, 4) + self.assertEqual(client.timeout, (10, 60)) + client.session.close() + + with self.assertRaises(TypeError): + WattTimeHistorical(None, None, False, 10, 4, (3, 5)) + + def test_timeout_error_message_points_at_knobs(self): + """A timeout is wrapped in RuntimeError with __cause__ preserved and a hint + naming both knobs -- including when urllib3's retry machinery has wrapped it + as ConnectionError -> MaxRetryError -> ReadTimeoutError, which is what the + session raises once its retries are exhausted.""" + url = "https://api.watttime.org/v3/test" + reason = urllib3.exceptions.ReadTimeoutError(None, url, "Read timed out.") + exhausted = requests.exceptions.ConnectionError( + urllib3.exceptions.MaxRetryError(None, url, reason=reason) + ) + + for exc in [requests.exceptions.ReadTimeout("too slow"), exhausted]: + with self.subTest(exc=type(exc).__name__): + base = self._authed_base(timeout=(10, 60)) + with mock.patch.object(base.session, "get", side_effect=exc): + with self.assertRaises(RuntimeError) as ctx: + base._make_rate_limited_request(url, {}) + + self.assertIs(ctx.exception.__cause__, exc) + message = str(ctx.exception) + self.assertIn("Hint", message) + self.assertIn("chunk_size", message) + self.assertIn("`timeout`", message) + self.assertIn("(10, 60)", message) + base.session.close() + + def test_non_timeout_error_message_has_no_hint(self): + base = self._authed_base() + error = requests.exceptions.HTTPError("403 Client Error: Forbidden") + + with mock.patch.object(base.session, "get", side_effect=error): + with self.assertRaises(RuntimeError) as ctx: + base._make_rate_limited_request("https://api.watttime.org/v3/test", {}) + + self.assertNotIn("Hint", str(ctx.exception)) + base.session.close() + @patch("time.sleep", return_value=None) def test_apply_rate_limit(self, mock_sleep): """Test _apply_rate_limit (single-threaded) triggers sleep when rate limit is exceeded.""" @@ -179,6 +295,34 @@ def test_get_chunks_rejects_inverted_span(self): with self.assertRaises(ValueError): self.base._get_chunks(start, end) + def test_get_chunks_honors_chunk_size(self): + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 31, tzinfo=UTC) + + chunks = self.base._get_chunks(start, end, chunk_size=timedelta(days=10)) + + self.assertEqual(len(chunks), 3) + self.assertEqual(chunks[0][0], start) + self.assertEqual(chunks[-1][1], end) + self.assertEqual(chunks[1][0] - chunks[0][1], timedelta(minutes=5)) + + # None means the 30-day default + self.assertEqual( + self.base._get_chunks(start, end, chunk_size=None), + self.base._get_chunks(start, end), + ) + + def test_get_chunks_rejects_chunk_size_at_or_below_trim(self): + # 5 minutes is trimmed from every chunk end but the last, so a chunk that + # short would be empty or inverted; zero or negative would never terminate. + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 2, tzinfo=UTC) + + for bad in [timedelta(minutes=5), timedelta(0), timedelta(minutes=-10)]: + with self.subTest(chunk_size=bad): + with self.assertRaises(ValueError): + self.base._get_chunks(start, end, chunk_size=bad) + @mock.patch("watttime.requests.Session.post", side_effect=mocked_register) def test_mock_register(self, mock_post): resp = self.base.register(email=os.getenv("WATTTIME_EMAIL")) @@ -202,6 +346,38 @@ def setUp(self): def tearDown(self): self.historical.session.close() + def test_get_historical_jsons_chunk_size_controls_request_count(self): + """30 days at 10-day chunks is three requests instead of one.""" + fake_response = {"data": [], "meta": {"model": {"date": "2024-01-01"}}} + + with mock.patch.object( + self.historical, "_fetch_data", return_value=[fake_response] + ) as mock_fetch: + self.historical.get_historical_jsons( + "2025-01-01 00:00Z", + "2025-01-31 00:00Z", + REGION, + chunk_size=timedelta(days=10), + ) + + param_chunks = mock_fetch.call_args.args[1] + self.assertEqual(len(param_chunks), 3) + + def test_get_historical_pandas_forwards_chunk_size(self): + fake_response = { + "data": [{"point_time": "2025-01-01T00:00:00+00:00", "value": 1.0}], + "meta": {}, + } + + with mock.patch.object( + self.historical, "get_historical_jsons", return_value=[fake_response] + ) as mock_jsons: + self.historical.get_historical_pandas( + "2025-01-01", "2025-01-02", REGION, chunk_size=timedelta(days=1) + ) + + self.assertEqual(mock_jsons.call_args.kwargs["chunk_size"], timedelta(days=1)) + def test_get_historical_jsons_3_months(self): start = "2025-01-01 00:00Z" end = "2025-03-31 00:00Z" diff --git a/watttime/api.py b/watttime/api.py index 7f04fae5..8884140a 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -18,6 +18,7 @@ from dateutil.parser import parse from pytz import UTC from requests.adapters import HTTPAdapter +import urllib3.exceptions from urllib3.util.retry import Retry try: @@ -27,6 +28,30 @@ VERSION = "0.0.0" +def _is_timeout(exc: BaseException) -> bool: + """ + True if `exc` is, or wraps, a connect/read timeout. + + Once the session's retries are exhausted, requests surfaces a read timeout as + ConnectionError -> MaxRetryError -> ReadTimeoutError rather than as ReadTimeout, + so the whole chain has to be walked, not just the top-level exception. + """ + seen = set() + stack = [exc] + while stack: + e = stack.pop() + if e is None or id(e) in seen: + continue + seen.add(id(e)) + if isinstance( + e, (requests.exceptions.Timeout, urllib3.exceptions.TimeoutError) + ): + return True + stack.extend([e.__cause__, e.__context__, getattr(e, "reason", None)]) + stack.extend(a for a in e.args if isinstance(a, BaseException)) + return False + + class WattTimeAPIWarning: def __init__(self, url: str, params: Dict[str, Any], warning_message: str): self.url = url @@ -71,6 +96,8 @@ def __init__( multithreaded: bool = False, rate_limit: int = 10, worker_count: int = min(10, (os.cpu_count() or 1) * 2), + *, + timeout: Optional[Union[float, Tuple[float, float]]] = (10, 60), ): """ Initializes a new instance of the class. @@ -81,6 +108,7 @@ def __init__( multithreaded (bool): Whether to use multithreading for requests. Default is False. rate_limit (int): The maximum number of requests to make per second. Default is 10 as this algins well with WattTime's API rate limiting policy. worker_count (int): The number of worker threads to use for multithreading. Default is min(10, (os.cpu_count() or 1) * 2). + timeout (Optional[Union[float, Tuple[float, float]]]): The timeout passed to every HTTP request, in seconds, using the standard `requests` forms: a single value applied to both the connect and read phases, or a (connect, read) tuple. `None` disables timeouts entirely. Default is (10, 60). Note that this is a per-attempt timeout: the session retries each request up to 3 times with backoff, so a request can take several times this long before raising. """ @@ -100,6 +128,7 @@ def __init__( self.multithreaded = multithreaded self.rate_limit = rate_limit + self.timeout = timeout self._last_request_times = [] self.worker_count = worker_count self.raised_warnings: List[WattTimeAPIWarning] = [] @@ -136,7 +165,7 @@ def _login(self): auth=requests.auth.HTTPBasicAuth( os.getenv("WATTTIME_USER"), os.getenv("WATTTIME_PASSWORD") ), - timeout=(10, 60), + timeout=self.timeout, ) rsp.raise_for_status() self.token = rsp.json().get("token", None) @@ -184,7 +213,10 @@ def _parse_dates( return start, end def _get_chunks( - self, start: datetime, end: datetime, chunk_size: timedelta = timedelta(days=30) + self, + start: datetime, + end: datetime, + chunk_size: Optional[timedelta] = None, ) -> List[Tuple[datetime, datetime]]: """ Generate a list of tuples representing chunks of time within a given time range. @@ -192,15 +224,23 @@ def _get_chunks( Args: start (datetime): The start datetime of the time range. end (datetime): The end datetime of the time range. - chunk_size (timedelta, optional): The size of each chunk. Defaults to timedelta(days=30). + chunk_size (Optional[timedelta], optional): The size of each chunk. None means the default of 30 days. + Must be longer than 5 minutes, since 5 minutes is trimmed from the end of every chunk but the last. Returns: List[Tuple[datetime, datetime]]: A list of tuples representing the chunks of time. If start == end, a single zero-length chunk is returned. Raises: - ValueError: If start is after end. + ValueError: If start is after end, or chunk_size is not longer than 5 minutes. """ + if chunk_size is None: + chunk_size = timedelta(days=30) + if chunk_size <= timedelta(minutes=5): + raise ValueError( + f"chunk_size must be longer than 5 minutes, got {chunk_size}" + ) + if start > end: raise ValueError(f"start ({start}) must not be after end ({end})") @@ -240,7 +280,7 @@ def register(self, email: str, organization: Optional[str] = None) -> None: "org": organization, } - rsp = self.session.post(url, json=params, timeout=(10, 60)) + rsp = self.session.post(url, json=params, timeout=self.timeout) rsp.raise_for_status() LOG.info( f"Successfully registered {os.getenv('WATTTIME_USER')}, please check {email} for a verification email" @@ -298,13 +338,23 @@ def _make_rate_limited_request(self, url: str, params: Dict[str, Any]) -> Dict: self._apply_rate_limit(ts) try: - rsp = self.session.get(url, headers=self.headers, params=params, timeout=60) + rsp = self.session.get( + url, headers=self.headers, params=params, timeout=self.timeout + ) rsp.raise_for_status() j = rsp.json() except requests.exceptions.RequestException as e: - raise RuntimeError( - f"API Request Failed: {e}\nURL: {url}\nParams: {params}" - ) from e + msg = f"API Request Failed: {e}\nURL: {url}\nParams: {params}" + if _is_timeout(e): + msg += ( + f"\nHint: the request exceeded the client timeout ({self.timeout}) " + "on every retry. This usually means the API was slow to respond, " + "which is more likely for requests covering a large time span. " + "Either pass a smaller `chunk_size` to the historical methods " + "(e.g. timedelta(days=10)) so each request covers less time, or " + "construct the client with a longer `timeout`." + ) + raise RuntimeError(msg) from e meta = j.get("meta", {}) warnings = meta.get("warnings") @@ -398,6 +448,8 @@ def get_historical_jsons( ] = "co2_moer", model: Optional[Union[str, date]] = None, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ) -> List[dict]: """ Base function to scrape historical data, returning a list of .json responses. @@ -409,6 +461,9 @@ def get_historical_jsons( signal_type (str, optional): one of ['co2_moer', 'co2_aoer', 'health_damage']. Defaults to "co2_moer". model (Optional[Union[str, date]], optional): Optionally provide a model, used for versioning models. Defaults to None. + chunk_size (Optional[timedelta], optional): The span of each request the date range is split into. + Defaults to 30 days. Smaller spans mean more, faster requests; use this when the API is + slow to answer large spans and requests are hitting the read timeout. Raises: Exception: Scraping failed for some reason @@ -423,7 +478,7 @@ def get_historical_jsons( params["include_imputed_marker"] = "true" start, end = self._parse_dates(start, end) - chunks = self._get_chunks(start, end) + chunks = self._get_chunks(start, end, chunk_size=chunk_size) # No model will default to the most recent model version available if model is not None: @@ -453,12 +508,14 @@ def get_historical_pandas( model: Optional[Union[str, date]] = None, include_meta: bool = False, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ): """ Return a pd.DataFrame with point_time, and values. Args: - See .get_hist_jsons() for shared arguments. + See .get_historical_jsons() for shared arguments. include_meta (bool, optional): adds additional columns to the output dataframe, containing the metadata information. Note that metadata is returned for each API response, not for each point_time. @@ -473,6 +530,7 @@ def get_historical_pandas( signal_type=signal_type, model=model, include_imputed_marker=include_imputed_marker, + chunk_size=chunk_size, ) df = pd.json_normalize( responses, record_path="data", meta=["meta"] if include_meta else [] @@ -492,6 +550,8 @@ def get_historical_csv( ] = "co2_moer", model: Optional[Union[str, date]] = None, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ): """ Retrieves historical data from a specified start date to an end date and saves it as a CSV file. @@ -503,6 +563,7 @@ def get_historical_csv( region (str): The region for which historical data is requested. signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]]): The type of signal for which historical data is requested. Default is "co2_moer". model (Optional[Union[str, date]]): The date of the model for which historical data is requested. It can be a string in the format "YYYY-MM-DD" or a date object. Default is None. + chunk_size (Optional[timedelta]): See .get_historical_jsons(). Default is None (30 days). Returns: None, results are saved to a csv file in the user's home directory. @@ -514,6 +575,7 @@ def get_historical_csv( signal_type=signal_type, model=model, include_imputed_marker=include_imputed_marker, + chunk_size=chunk_size, ) out_dir = Path.home() / "watttime_historical_csvs" From ed4f2c498eb24b3e040f683cba4d8a510b5d3ff8 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:19:01 -0400 Subject: [PATCH 3/5] Document timeout and chunk_size in the README Add a short "Tuning requests that time out" note under the historical data example showing both options and that the timeout is per attempt. Co-Authored-By: Claude Fable 5.1 --- README.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/README.md b/README.md index 6a680742..11faf371 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,29 @@ wt_hist.get_historical_csv( ) ``` +#### Tuning requests that time out + +Historical pulls are split into 30-day requests, each with a 60 s read timeout. If the API is slow to answer a large span, a request can time out and the error message will say so. Two options, which can be combined: + +```python +from datetime import timedelta +from watttime import WattTimeHistorical + +# allow each request more time: (connect, read) seconds +wt_hist = WattTimeHistorical(username, password, timeout=(10, 300)) + +# or make each request cover less time +moers = wt_hist.get_historical_pandas( + start = '2022-01-01 00:00Z', + end = '2023-01-01 00:00Z', + region = 'CAISO_NORTH', + signal_type = 'co2_moer', + chunk_size = timedelta(days=10) +) +``` + +The timeout applies to each attempt; the client retries a failed request up to three times before raising. + You could also combine these classes to iterate through all regions where you have access to data: ```python From 266eca1793171de060c2f1c9031c9283cdd6aa64 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:39:59 -0400 Subject: [PATCH 4/5] Retry by failure class: split slow span requests instead of repeating them The transport retried every failure identically, up to four times. For a read timeout on a 30-day request that multiplies load on an already slow backend and turns a one-minute failure into a four-minute one. Split the retry responsibility in two: - Transport (urllib3 Retry) keeps connection errors, 500/502/503, and now 429 (honouring Retry-After). read=0: read timeouts are never retried as-is, and 504 leaves the forcelist -- both mean the API was too slow to answer this much data, which repeating does not fix. - Client: _fetch_adaptive halves a span request's time range on a timeout-class failure and recurses, up to _MAX_TIMEOUT_SPLITS = 2 levels, depth-first and failing fast. A 401 on a token the client considered valid triggers one serialised re-login and one retry. Data-path failures are now WattTimeRequestError, a RuntimeError subclass carrying `kind`, `url` and `params`; existing `except RuntimeError` handlers and the message format are unchanged. Every split and re-login is logged at WARNING so a run that succeeds by adapting still records that the API was slow. Co-Authored-By: Claude Fable 5.1 --- tests/test_sdk.py | 235 ++++++++++++++++++++++++++++++++++++++++++++++ watttime/api.py | 199 ++++++++++++++++++++++++++++++++------- 2 files changed, 399 insertions(+), 35 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 9db18a42..5132ddd8 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -8,6 +8,7 @@ import os from watttime import ( WattTimeBase, + WattTimeRequestError, WattTimeHistorical, WattTimeMyAccess, WattTimeForecast, @@ -190,6 +191,240 @@ def test_non_timeout_error_message_has_no_hint(self): self.assertNotIn("Hint", str(ctx.exception)) base.session.close() + # --- failure-class-aware retries --- + + @staticmethod + def _http_error(status): + return requests.exceptions.HTTPError( + f"{status} error", response=mock.Mock(status_code=status) + ) + + @staticmethod + def _response_raising(status): + rsp = mock.Mock() + rsp.raise_for_status.side_effect = TestWattTimeBase._http_error(status) + return rsp + + @staticmethod + def _ok_response(): + rsp = mock.Mock() + rsp.json.return_value = {"data": [], "meta": {}} + rsp.raise_for_status.return_value = None + return rsp + + @staticmethod + def _slow_api(max_ok_span, calls): + """A stand-in for _make_rate_limited_request that times out on spans longer + than max_ok_span and records every call.""" + + def fake(url, params): + calls.append(params) + start, end = params.get("start"), params.get("end") + if not (isinstance(start, datetime) and isinstance(end, datetime)): + raise WattTimeRequestError("no span", "timeout", url, params) + if end - start > max_ok_span: + raise WattTimeRequestError("timed out", "timeout", url, params) + return {"data": [], "meta": {}} + + return fake + + def test_transport_retry_policy(self): + """The transport retries connection errors, transient 5xx and 429 -- never + read timeouts or 504, which the client handles by asking for less at once.""" + retry = self.base.session.get_adapter("https://api.watttime.org").max_retries + self.assertEqual(retry.total, 3) + self.assertEqual(retry.read, 0) + self.assertIn(429, retry.status_forcelist) + self.assertNotIn(504, retry.status_forcelist) + + def test_request_error_kind_classification(self): + url = "https://api.watttime.org/v3/test" + exhausted = requests.exceptions.ConnectionError( + urllib3.exceptions.MaxRetryError( + None, + url, + reason=urllib3.exceptions.ReadTimeoutError( + None, url, "Read timed out." + ), + ) + ) + cases = [ + (requests.exceptions.ReadTimeout("slow"), "timeout"), + (exhausted, "timeout"), + (self._http_error(504), "timeout"), + (self._http_error(401), "http_401"), + (self._http_error(403), "http_4xx"), + (self._http_error(500), "http_5xx"), + (requests.exceptions.ConnectionError("reset by peer"), "connection"), + (requests.exceptions.HTTPError("no response attached"), "other"), + ] + for exc, kind in cases: + with self.subTest(kind=kind, exc=type(exc).__name__): + base = self._authed_base() + fake_login = lambda: setattr(base, "headers", {"Authorization": "new"}) + with mock.patch.object(base, "_login", side_effect=fake_login): + with mock.patch.object(base.session, "get", side_effect=exc): + with self.assertRaises(WattTimeRequestError) as ctx: + base._make_rate_limited_request(url, {}) + self.assertEqual(ctx.exception.kind, kind) + self.assertIsInstance(ctx.exception, RuntimeError) + base.session.close() + + def test_401_refreshes_token_and_retries_once(self): + base = self._authed_base() + + def fake_login(): + base.headers = {"Authorization": "Bearer refreshed"} + base.token_valid_until = datetime.now() + timedelta(minutes=30) + + with mock.patch.object(base, "_login", side_effect=fake_login) as mock_login: + with mock.patch.object( + base.session, + "get", + side_effect=[self._response_raising(401), self._ok_response()], + ) as mock_get: + result = base._make_rate_limited_request( + "https://api.watttime.org/v3/test", {} + ) + + self.assertEqual(result, {"data": [], "meta": {}}) + mock_login.assert_called_once() + self.assertEqual(mock_get.call_count, 2) + self.assertEqual( + mock_get.call_args.kwargs["headers"], {"Authorization": "Bearer refreshed"} + ) + base.session.close() + + def test_second_401_is_an_auth_failure(self): + base = self._authed_base() + fake_login = lambda: setattr(base, "headers", {"Authorization": "refreshed"}) + + with mock.patch.object(base, "_login", side_effect=fake_login) as mock_login: + with mock.patch.object( + base.session, "get", return_value=self._response_raising(401) + ) as mock_get: + with self.assertRaises(WattTimeRequestError) as ctx: + base._make_rate_limited_request( + "https://api.watttime.org/v3/test", {} + ) + + self.assertEqual(ctx.exception.kind, "http_401") + mock_login.assert_called_once() + self.assertEqual(mock_get.call_count, 2) + base.session.close() + + def test_other_4xx_is_not_retried(self): + base = self._authed_base() + + with mock.patch.object(base, "_login") as mock_login: + with mock.patch.object( + base.session, "get", return_value=self._response_raising(403) + ) as mock_get: + with self.assertRaises(WattTimeRequestError) as ctx: + base._make_rate_limited_request( + "https://api.watttime.org/v3/test", {} + ) + + self.assertEqual(ctx.exception.kind, "http_4xx") + mock_login.assert_not_called() + self.assertEqual(mock_get.call_count, 1) + base.session.close() + + def test_timeout_on_span_request_splits_and_retries(self): + """A 30-day chunk the API cannot answer in time becomes two 15-day requests.""" + base = self._authed_base() + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 31, tzinfo=UTC) + calls = [] + + with mock.patch.object( + base, + "_make_rate_limited_request", + side_effect=self._slow_api(timedelta(days=15), calls), + ): + with mock.patch("watttime.api.LOG.warning") as mock_warning: + responses = base._fetch_data( + "https://x/v3/historical", + [{"region": REGION, "start": start, "end": end}], + ) + + self.assertEqual(len(responses), 2) + self.assertEqual(len(calls), 3) # one timeout, then the two halves + self.assertEqual(calls[1]["start"], start) + self.assertEqual(calls[2]["end"], end) + self.assertEqual(calls[1]["region"], REGION) + # halves must not overlap, same as the chunker's own rule + self.assertEqual(calls[2]["start"] - calls[1]["end"], timedelta(minutes=5)) + mock_warning.assert_called_once() + self.assertIn("splitting", mock_warning.call_args.args[0]) + base.session.close() + + def test_split_stops_at_depth_cap(self): + """Depth-first: the first sub-span that still times out at the last level + fails the call, so a dead API costs one timeout per level, not one per leaf.""" + base = self._authed_base() + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 31, tzinfo=UTC) + calls = [] + + with mock.patch.object( + base, + "_make_rate_limited_request", + side_effect=self._slow_api(timedelta(0), calls), + ): + with mock.patch("watttime.api.LOG.warning") as mock_warning: + with self.assertRaises(WattTimeRequestError) as ctx: + base._fetch_adaptive("https://x", {"start": start, "end": end}) + + self.assertEqual(ctx.exception.kind, "timeout") + self.assertEqual(len(calls), 1 + base._MAX_TIMEOUT_SPLITS) + self.assertEqual(mock_warning.call_count, base._MAX_TIMEOUT_SPLITS) + base.session.close() + + def test_no_split_for_non_span_requests(self): + """Only requests with datetime start/end are split; anything else fails once.""" + base = self._authed_base() + calls = [] + fake = self._slow_api(timedelta(0), calls) + + for params in [ + {}, + {"start": "2025-01-01T00:00Z", "end": "2025-01-02T00:00Z"}, + ]: + with self.subTest(params=params): + calls.clear() + with mock.patch.object( + base, "_make_rate_limited_request", side_effect=fake + ): + with self.assertRaises(WattTimeRequestError): + base._fetch_adaptive("https://x", params) + self.assertEqual(len(calls), 1) + base.session.close() + + def test_multithreaded_fetch_splits_inside_the_worker(self): + base = self._authed_base(multithreaded=True, worker_count=2) + start = datetime(2025, 1, 1, tzinfo=UTC) + chunks = [ + { + "start": start + timedelta(days=30 * i), + "end": start + timedelta(days=30 * (i + 1)), + } + for i in range(2) + ] + calls = [] + + with mock.patch.object( + base, + "_make_rate_limited_request", + side_effect=self._slow_api(timedelta(days=15), calls), + ): + with mock.patch("watttime.api.LOG.warning"): + responses = base._fetch_data("https://x", chunks) + + self.assertEqual(len(responses), 4) + self.assertEqual(len(calls), 6) # 2 timeouts + 4 halves + base.session.close() + @patch("time.sleep", return_value=None) def test_apply_rate_limit(self, mock_sleep): """Test _apply_rate_limit (single-threaded) triggers sleep when rate limit is exceeded.""" diff --git a/watttime/api.py b/watttime/api.py index 8884140a..fa9fc4e0 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -52,6 +52,45 @@ def _is_timeout(exc: BaseException) -> bool: return False +class WattTimeRequestError(RuntimeError): + """ + Raised when an API request fails after the client has done what it can about it. + + Subclasses RuntimeError so existing `except RuntimeError` handlers keep working. + `kind` names the failure class so callers can branch on it instead of parsing + the message: "timeout", "http_401", "http_4xx", "http_5xx", "connection" or "other". + """ + + def __init__(self, message: str, kind: str, url: str, params: Dict[str, Any]): + super().__init__(message) + self.kind = kind + self.url = url + self.params = params + + +def _classify_request_exception(exc: requests.exceptions.RequestException) -> str: + """ + Sort a failed request into the class that decides how the client responds. + + A 504 is grouped with read timeouts: the gateway gave up waiting on the same + slow upstream query, so it calls for the same response (ask for less at once). + """ + if _is_timeout(exc): + return "timeout" + status = getattr(getattr(exc, "response", None), "status_code", None) + if status == 504: + return "timeout" + if status == 401: + return "http_401" + if status is not None and 400 <= status < 500: + return "http_4xx" + if status is not None and 500 <= status < 600: + return "http_5xx" + if isinstance(exc, requests.exceptions.ConnectionError): + return "connection" + return "other" + + class WattTimeAPIWarning: def __init__(self, url: str, params: Dict[str, Any], warning_message: str): self.url = url @@ -108,7 +147,7 @@ def __init__( multithreaded (bool): Whether to use multithreading for requests. Default is False. rate_limit (int): The maximum number of requests to make per second. Default is 10 as this algins well with WattTime's API rate limiting policy. worker_count (int): The number of worker threads to use for multithreading. Default is min(10, (os.cpu_count() or 1) * 2). - timeout (Optional[Union[float, Tuple[float, float]]]): The timeout passed to every HTTP request, in seconds, using the standard `requests` forms: a single value applied to both the connect and read phases, or a (connect, read) tuple. `None` disables timeouts entirely. Default is (10, 60). Note that this is a per-attempt timeout: the session retries each request up to 3 times with backoff, so a request can take several times this long before raising. + timeout (Optional[Union[float, Tuple[float, float]]]): The timeout passed to every HTTP request, in seconds, using the standard `requests` forms: a single value applied to both the connect and read phases, or a (connect, read) tuple. `None` disables timeouts entirely. Default is (10, 60). A request that times out is not retried as-is; on the chunked endpoints its time span is split in half and retried (up to twice), and elsewhere it fails with a hint. Connection errors and transient server errors are retried up to 3 times with backoff. """ @@ -132,6 +171,7 @@ def __init__( self._last_request_times = [] self.worker_count = worker_count self.raised_warnings: List[WattTimeAPIWarning] = [] + self._login_lock = threading.Lock() if self.multithreaded: self._rate_limit_lock = ( @@ -139,9 +179,16 @@ def __init__( ) # prevent multiple threads from modifying _last_request_times simultaneously self._rate_limit_condition = threading.Condition(self._rate_limit_lock) + # The transport retries what re-issuing the identical request can fix: + # connection errors, transient 5xx, and 429 (honouring Retry-After). + # It does NOT retry read timeouts or 504: those mean the API was slow to + # answer *this much* data, so the client responds by asking for less at a + # time instead (see _fetch_adaptive). Repeating a slow query four times + # only multiplies the load on a backend that is already struggling. retry_strategy = Retry( total=3, - status_forcelist=[500, 502, 503, 504], + read=0, + status_forcelist=[429, 500, 502, 503], backoff_factor=1, raise_on_status=False, ) @@ -317,16 +364,28 @@ def region_from_loc( j = self._make_rate_limited_request(url, params=params) return j - def _make_rate_limited_request(self, url: str, params: Dict[str, Any]) -> Dict: + def _ensure_logged_in(self): """ - Makes a single API request while respecting the rate limit. + Log in if there is no token or the local 30-minute clock has run out. + Serialised so that concurrent workers reaching an expired token log in once. """ + if self._is_token_valid() and self.headers: + return + with self._login_lock: + if not self._is_token_valid() or not self.headers: + self._login() - # should already be logged in -- keeping incase long running chunked request surpasses - # token timeout - if not self._is_token_valid() or not self.headers: - self._login() + def _refresh_token(self, rejected_headers: Optional[Dict[str, str]]): + """ + Log in again after the API rejected a token the client thought was valid. + Workers that all saw the same rejected token refresh it once, not once each. + """ + with self._login_lock: + if self.headers is rejected_headers: + self._login() + def _rate_limited_get_json(self, url: str, params: Dict[str, Any]) -> Dict: + """One GET, rate limited, raising requests exceptions untouched.""" ts = time.time() # apply rate limiting by either sleeping (single thread) or @@ -337,24 +396,53 @@ def _make_rate_limited_request(self, url: str, params: Dict[str, Any]) -> Dict: else: self._apply_rate_limit(ts) - try: - rsp = self.session.get( - url, headers=self.headers, params=params, timeout=self.timeout + rsp = self.session.get( + url, headers=self.headers, params=params, timeout=self.timeout + ) + rsp.raise_for_status() + return rsp.json() + + def _request_error( + self, exc: requests.exceptions.RequestException, url: str, params: Dict + ) -> WattTimeRequestError: + kind = _classify_request_exception(exc) + msg = f"API Request Failed: {exc}\nURL: {url}\nParams: {params}" + if kind == "timeout": + msg += ( + f"\nHint: the request exceeded the client timeout ({self.timeout}). " + "This usually means the API was slow to respond, which is more likely " + "for requests covering a large time span. Either pass a smaller " + "`chunk_size` to the historical methods (e.g. timedelta(days=10)) so " + "each request covers less time, or construct the client with a longer " + "`timeout`." ) - rsp.raise_for_status() - j = rsp.json() - except requests.exceptions.RequestException as e: - msg = f"API Request Failed: {e}\nURL: {url}\nParams: {params}" - if _is_timeout(e): - msg += ( - f"\nHint: the request exceeded the client timeout ({self.timeout}) " - "on every retry. This usually means the API was slow to respond, " - "which is more likely for requests covering a large time span. " - "Either pass a smaller `chunk_size` to the historical methods " - "(e.g. timedelta(days=10)) so each request covers less time, or " - "construct the client with a longer `timeout`." + return WattTimeRequestError(msg, kind=kind, url=url, params=params) + + def _make_rate_limited_request(self, url: str, params: Dict[str, Any]) -> Dict: + """ + Makes a single API request while respecting the rate limit. + + A 401 on a token the client still considers valid is refreshed and retried + once; a second 401 is a real authentication failure. Every other failure is + raised as WattTimeRequestError with its `kind` set. + """ + self._ensure_logged_in() + headers_used = self.headers + + try: + try: + j = self._rate_limited_get_json(url, params) + except requests.exceptions.HTTPError as e: + if _classify_request_exception(e) != "http_401": + raise + LOG.warning( + f"API returned 401 for a token the client considered valid; " + f"refreshing and retrying once | URL: {url} | Params: {params}" ) - raise RuntimeError(msg) from e + self._refresh_token(headers_used) + j = self._rate_limited_get_json(url, params) + except requests.exceptions.RequestException as e: + raise self._request_error(e, url, params) from e meta = j.get("meta", {}) warnings = meta.get("warnings") @@ -398,21 +486,65 @@ def _apply_rate_limit(self, ts: float): if self.multithreaded: self._rate_limit_condition.notify_all() + # How many times one chunk may be halved after timing out. A chunk becomes at + # most 2**_MAX_TIMEOUT_SPLITS requests. Splitting is depth-first and the first + # sub-span that still times out at the last level fails the call, so a chunk + # the API never answers costs about (1 + _MAX_TIMEOUT_SPLITS) x read timeout. + _MAX_TIMEOUT_SPLITS = 2 + + def _fetch_adaptive( + self, url: str, params: Dict[str, Any], depth: int = 0 + ) -> List[Dict]: + """ + Fetch one set of params, halving its time span and retrying on timeout. + + A read timeout or 504 on a span request usually means the API was slow to + answer that much data at once, so re-issuing the identical request tends to + fail the same way. Asking for half the span instead is what actually helps. + Only applies when `start` and `end` are datetimes (the chunked endpoints), and + at most _MAX_TIMEOUT_SPLITS times per original chunk. Every split is logged. + """ + try: + return [self._make_rate_limited_request(url, params)] + except WattTimeRequestError as e: + start, end = params.get("start"), params.get("end") + if not ( + e.kind == "timeout" + and isinstance(start, datetime) + and isinstance(end, datetime) + and depth < self._MAX_TIMEOUT_SPLITS + ): + raise + half = (end - start) / 2 + if half <= timedelta(minutes=5): # the chunker's floor + raise + halves = self._get_chunks(start, end, chunk_size=half) + LOG.warning( + f"Request timed out (timeout={self.timeout}); splitting {start} -> {end} " + f"into {len(halves)} requests of {half} and retrying " + f"(split {depth + 1} of {self._MAX_TIMEOUT_SPLITS}) | URL: {url}" + ) + + responses = [] + for sub_start, sub_end in halves: + sub_params = {**params, "start": sub_start, "end": sub_end} + responses.extend(self._fetch_adaptive(url, sub_params, depth + 1)) + return responses + def _fetch_data( self, url: str, param_chunks: Union[Dict[str, Any], List[Dict[str, Any]]], ) -> List[Dict]: """ - Base method for fetching data without multithreading. + Fetch a series of requests with varying `param_chunks`, sequentially or with + a thread pool. Each chunk goes through _fetch_adaptive, so a chunk that times + out is split into smaller spans rather than failing the whole call. If you are making a single request, you can call _make_rate_limited_request directly. - This class is suited for making a series of requests in a for loop, with - varying `param_chunks`. """ # first try to login before beginning multithreading - if not self._is_token_valid() or not self.headers: - self._login() + self._ensure_logged_in() if isinstance(param_chunks, dict): param_chunks = [param_chunks] @@ -421,18 +553,15 @@ def _fetch_data( if self.multithreaded: with ThreadPoolExecutor(max_workers=self.worker_count) as executor: futures = { - executor.submit( - self._make_rate_limited_request, url, params - ): params + executor.submit(self._fetch_adaptive, url, params): params for params in param_chunks } for future in as_completed(futures): - responses.append(future.result()) + responses.extend(future.result()) else: for params in param_chunks: - rsp = self._make_rate_limited_request(url, params) - responses.append(rsp) + responses.extend(self._fetch_adaptive(url, params)) return responses From 54b328888bc320b0502db0eaf915271f67f6c350 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:40:00 -0400 Subject: [PATCH 5/5] Document adaptive retries and WattTimeRequestError in the README Co-Authored-By: Claude Fable 5.1 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 11faf371..f782af25 100644 --- a/README.md +++ b/README.md @@ -95,7 +95,7 @@ moers = wt_hist.get_historical_pandas( ) ``` -The timeout applies to each attempt; the client retries a failed request up to three times before raising. +A request that times out is not simply retried: on the historical endpoints the client splits its time span in half and asks again (up to twice), logging a warning each time, so occasional slowness is absorbed without any of the above. Connection errors and transient server errors are retried up to three times with backoff. When the client does give up it raises `WattTimeRequestError`, whose `kind` attribute (`"timeout"`, `"http_401"`, `"http_4xx"`, `"http_5xx"`, `"connection"`) lets your code decide what to do without parsing the message. You could also combine these classes to iterate through all regions where you have access to data: