diff --git a/README.md b/README.md
index cf06d3ac..95a35865 100644
--- a/README.md
+++ b/README.md
@@ -38,6 +38,47 @@ cwms.init_session(
If both `token` and `api_key` are provided, `cwms-python` will use the token
and log a warning.
+### HTTP connection pools
+
+Sessions created with `cwms.init_session(api_root=..., pool_connections=100)`
+configure the same connection pool size and retry policy for HTTP and HTTPS.
+This includes unencrypted internal HTTP CDA roots used by Batch jobs. The default
+pool retains up to 100 connections per host for reuse; it does not cap concurrent
+requests. Set `pool_connections` when creating the session to size that pool for
+your workload. A `Connection pool is full, discarding connection` warning means
+an extra connection is being closed instead of retained, not that a response or
+its data was discarded. Check raised exceptions for actual request failures.
+
+### Errors and debugging
+
+Failed HTTP requests raise `cwms.api.ApiError`. Its message includes the HTTP
+status, method, URL, and CDA response body (including incident details when
+provided). The original response is available as `error.response`. Custom
+user-management errors retain these details too. Network exceptions propagate
+with their original type. Invalid JSON responses raise `ApiError` with the
+decoding exception as the cause; empty response bodies return an empty dictionary.
+
+Concurrent time-series reads and writes raise `cwms.api.BatchError` if any
+series or chunk fails. It subclasses `RuntimeError`, and `error.failures`
+contains `(series_or_chunk, original_exception)` pairs for every failure.
+Reads do not return incomplete results as success. Successful writes are not
+rolled back. Failures while looking up time-series extents also propagate.
+Chunk retries are limited to connection errors, timeouts, and HTTP
+429/500/502/503/504; validation and other permanent errors fail immediately
+at the chunk layer. The shared HTTP adapter retains its existing retry policy.
+
+Enable request outcome and chunk diagnostics with Python logging:
+
+```python
+import logging
+
+logging.basicConfig(level=logging.WARNING)
+logging.getLogger("cwms").setLevel(logging.DEBUG)
+```
+
+Request diagnostics include the method, endpoint, and response status, without
+request bodies or authentication headers.
+
## Getting Started
```python
diff --git a/cwms/api.py b/cwms/api.py
index 20d2c2c5..6c710d09 100644
--- a/cwms/api.py
+++ b/cwms/api.py
@@ -33,13 +33,14 @@
import json
import logging
from http import HTTPStatus
-from json import JSONDecodeError
-from typing import Any, Optional, cast
+from typing import Any, Optional, Union, cast
from requests import Response, adapters
+from requests.exceptions import JSONDecodeError, RequestException
from requests.exceptions import RetryError as RequestsRetryError
from requests_toolbelt import sessions # type: ignore
from requests_toolbelt.sessions import BaseUrlSession # type: ignore
+from urllib3.exceptions import HTTPError as Urllib3HTTPError
from urllib3.util.retry import Retry
from cwms.cwms_types import JSON, RequestParams
@@ -47,6 +48,7 @@
# Specify the default API root URL and version.
API_ROOT = "https://cwms-data.usace.army.mil/cwms-data/"
API_VERSION = 2
+logger = logging.getLogger(__name__)
# Specify whether LRTS will use new ID format
USE_NEW_LRTS_IDS = False
@@ -71,6 +73,7 @@
pool_connections=100, pool_maxsize=100, max_retries=retry_strategy
)
SESSION.mount("https://", adapter)
+SESSION.mount("http://", adapter)
class InvalidVersion(Exception):
@@ -90,17 +93,19 @@ def __init__(self, response: Response, message: Optional[str] = None):
self.message = message
def __str__(self) -> str:
- if self.message:
- return self.message
-
# Include the request URL in the error message.
- message = f"CWMS API Error ({self.response.url})"
+ message = f"CWMS API Error ({self.response.url}) {self.response.status_code}"
+ request = getattr(self.response, "request", None)
+ if request is not None:
+ message += f" {request.method}"
# If a reason is provided in the response, add it to the message.
if reason := self.response.reason:
message += f" {reason}"
message += "."
+ if self.message:
+ message += f" {self.message}"
# Add additional context to help the user resolve the issue.
hint = self.hint()
@@ -111,10 +116,7 @@ def __str__(self) -> str:
content = getattr(self.response, "content", None)
if content:
if isinstance(content, bytes):
- try:
- text = content.decode("utf-8", errors="replace")
- except Exception:
- text = repr(content)
+ text = content.decode("utf-8", errors="replace")
else:
text = str(content)
message += f" {text}"
@@ -130,6 +132,9 @@ def hint(self) -> str:
if status == 400:
return "Check that your parameters are correct."
if status == 404:
+ request = getattr(self.response, "request", None)
+ if request is not None and request.method != "GET":
+ return "Check the CDA response body for the failed operation."
return "May be the result of an empty query."
# No hint for other codes
@@ -144,8 +149,29 @@ class PermissionError(ApiError):
"""Raised when the CDA request is not authorized for the current caller."""
-def _unwrap_retry_error(error: RequestsRetryError) -> Exception:
- """Return the original retry cause when requests wraps it in RetryError."""
+class BatchError(RuntimeError):
+ """A concurrent operation failed; ``failures`` retains every original exception.
+
+ Each failure is a (series or chunk description, exception) pair. Successful
+ writes are not rolled back. This remains compatible with RuntimeError handlers.
+ """
+
+ def __init__(self, message: str, failures: list[tuple[str, Exception]]):
+ self.failures = failures
+ super().__init__(
+ message
+ + "\n"
+ + "\n".join(f"{context}: {error}" for context, error in failures)
+ )
+
+
+def _unwrap_retry_error(
+ error: RequestsRetryError,
+) -> Union[RequestException, Urllib3HTTPError]:
+ """Unwrap transport errors; retain RetryError for other causes.
+
+ Unknown causes remain available through the original wrapper's chain or args.
+ """
current: Exception = error
cause = error.__cause__
@@ -162,7 +188,9 @@ def _unwrap_retry_error(error: RequestsRetryError) -> Exception:
current = reason
reason = getattr(current, "reason", None)
- return current
+ if isinstance(current, (RequestException, Urllib3HTTPError)):
+ return current
+ return error
def init_session(
@@ -185,6 +213,9 @@ def init_session(
api_key (optional): An authentication key.
token (optional): A Keycloak access token. If both token and api_key are
provided, token is used.
+ pool_connections (optional): Number of host pools and reusable connections
+ per host when creating a session with api_root. Defaults to 100 for
+ both HTTP and HTTPS; this is not a limit on concurrent requests.
Returns:
Returns the updated session object.
@@ -202,6 +233,7 @@ def init_session(
max_retries=retry_strategy,
)
SESSION.mount("https://", adapter)
+ SESSION.mount("http://", adapter)
if token:
if api_key:
logging.warning(
@@ -336,6 +368,8 @@ def get_xml(
def _process_response(response: Response) -> Any:
+ if not response.content:
+ return {}
try:
# Avoid case sensitivity issues with the content type header
content_type = response.headers.get("Content-Type", "").lower()
@@ -356,10 +390,14 @@ def _process_response(response: Response) -> Any:
# Fallback for remaining content types
return response.content.decode("utf-8")
except JSONDecodeError as error:
- logging.error(
- f"Error decoding CDA response as JSON: {error} on line {error.lineno}\n\tFalling back to text"
- )
- return response.text
+ raise ApiError(response, "Invalid JSON in CDA response.") from error
+
+
+def _check_response(response: Response, method: str, endpoint: str) -> None:
+ """Record request outcomes without logging credentials or request bodies."""
+ logger.debug("CDA %s %s returned HTTP %s", method, endpoint, response.status_code)
+ if not response.ok:
+ raise ApiError(response)
def get(
@@ -391,12 +429,16 @@ def get(
}
try:
with SESSION.get(endpoint, params=params, headers=headers) as response:
- if not response.ok:
- logging.error(f"CDA Error: response={response}")
- raise ApiError(response)
+ _check_response(response, "GET", endpoint)
return _process_response(response)
except RequestsRetryError as error:
- raise _unwrap_retry_error(error) from None
+ cause = _unwrap_retry_error(error)
+ if cause is error:
+ raise
+ raise cause from error
+ except RequestException as error:
+ logger.debug("CDA GET %s failed: %s", endpoint, type(error).__name__)
+ raise
def get_with_paging(
@@ -459,12 +501,16 @@ def _post_function(
with SESSION.post(
endpoint, params=params, headers=headers, data=data
) as response:
- if not response.ok:
- logging.error(f"CDA Error: response={response}")
- raise ApiError(response)
+ _check_response(response, "POST", endpoint)
return response
except RequestsRetryError as error:
- raise _unwrap_retry_error(error) from None
+ cause = _unwrap_retry_error(error)
+ if cause is error:
+ raise
+ raise cause from error
+ except RequestException as error:
+ logger.debug("CDA POST %s failed: %s", endpoint, type(error).__name__)
+ raise
def post(
@@ -556,17 +602,21 @@ def patch(
"X-CWMS-LRTS-Formatting": str(USE_NEW_LRTS_IDS).lower(),
}
- if data and isinstance(data, dict) or isinstance(data, list):
+ if isinstance(data, (dict, list)):
data = json.dumps(data)
try:
with SESSION.patch(
endpoint, params=params, headers=headers, data=data
) as response:
- if not response.ok:
- logging.error(f"CDA Error: response={response}")
- raise ApiError(response)
+ _check_response(response, "PATCH", endpoint)
except RequestsRetryError as error:
- raise _unwrap_retry_error(error) from None
+ cause = _unwrap_retry_error(error)
+ if cause is error:
+ raise
+ raise cause from error
+ except RequestException as error:
+ logger.debug("CDA PATCH %s failed: %s", endpoint, type(error).__name__)
+ raise
def delete(
@@ -574,6 +624,7 @@ def delete(
params: Optional[RequestParams] = None,
*,
api_version: int = API_VERSION,
+ data: Optional[Any] = None,
) -> None:
"""Make a DELETE request to the CWMS Data API.
@@ -584,6 +635,7 @@ def delete(
Keyword Args:
api_version (optional): The CDA version to use for the request. If not specified,
the default API_VERSION will be used.
+ data (optional): Request body, JSON-encoded for dictionaries and lists.
Raises:
ApiError: If an error response is return by the API.
@@ -593,10 +645,20 @@ def delete(
"Accept": api_version_text(api_version),
"X-CWMS-LRTS-Formatting": str(USE_NEW_LRTS_IDS).lower(),
}
+ kwargs: dict[str, Any] = {}
+ if data is not None:
+ headers["Content-Type"] = api_version_text(api_version)
+ kwargs["data"] = json.dumps(data) if isinstance(data, (dict, list)) else data
try:
- with SESSION.delete(endpoint, params=params, headers=headers) as response:
- if not response.ok:
- logging.error(f"CDA Error: response={response}")
- raise ApiError(response)
+ with SESSION.delete(
+ endpoint, params=params, headers=headers, **kwargs
+ ) as response:
+ _check_response(response, "DELETE", endpoint)
except RequestsRetryError as error:
- raise _unwrap_retry_error(error) from None
+ cause = _unwrap_retry_error(error)
+ if cause is error:
+ raise
+ raise cause from error
+ except RequestException as error:
+ logger.debug("CDA DELETE %s failed: %s", endpoint, type(error).__name__)
+ raise
diff --git a/cwms/ratings/ratings.py b/cwms/ratings/ratings.py
index b6c2339f..936f1857 100644
--- a/cwms/ratings/ratings.py
+++ b/cwms/ratings/ratings.py
@@ -424,8 +424,8 @@ def _validate_rating_params(
raise ValueError(f"Invalid rating identifer: {rating_id}")
try:
ind_params, _ = parts[1].split(";")
- except Exception:
- raise ValueError(f"Invalid rating template: {parts[1]}")
+ except ValueError as error:
+ raise ValueError(f"Invalid rating template: {parts[1]}") from error
if not office_id:
raise ValueError("Cannot rate values without an office identifier")
if not units:
@@ -466,8 +466,8 @@ def _perform_value_rating(
ind_params = _validate_rating_params(rating_id, office_id, units, values)
try:
ind_units_str, dep_unit = units.split(";")
- except Exception:
- raise ValueError("Invalid units string")
+ except ValueError as error:
+ raise ValueError("Invalid units string") from error
value_count = len(values[0])
times = _get_times(value_count, times)
if not rating_time:
diff --git a/cwms/ratings/ratings_spec.py b/cwms/ratings/ratings_spec.py
index c482c66f..1856dd67 100644
--- a/cwms/ratings/ratings_spec.py
+++ b/cwms/ratings/ratings_spec.py
@@ -127,7 +127,7 @@ def rating_spec_df_to_xml(data: pd.DataFrame) -> str:
try:
spec_xml += f"""
{data.loc[0,'source-agency']}"""
- except Exception:
+ except KeyError:
spec_xml += """
"""
spec_xml += f"""
@@ -154,7 +154,7 @@ def rating_spec_df_to_xml(data: pd.DataFrame) -> str:
try:
spec_xml2 += f"""
{data.loc[0,'description']}"""
- except Exception:
+ except KeyError:
spec_xml2 += """
"""
spec_xml2 += """
diff --git a/cwms/timeseries/timeseries.py b/cwms/timeseries/timeseries.py
index 33d5fb1a..b93170c5 100644
--- a/cwms/timeseries/timeseries.py
+++ b/cwms/timeseries/timeseries.py
@@ -6,11 +6,14 @@
import pandas as pd
from pandas import DataFrame
+from requests.exceptions import ConnectionError, Timeout
import cwms.api as api
from cwms.catalog.catalog import get_ts_extents
from cwms.cwms_types import JSON, Data
+logger = logging.getLogger(__name__)
+
_DEFAULT_CHUNK_DAYS = 365
_MIN_INTERVAL_MINUTES = 2
_FINE_INTERVAL_MINUTES = 15
@@ -129,45 +132,53 @@ def get_multi_timeseries_df(
"""
def get_ts_ids(ts_id: str) -> Any:
- try:
- if ":" in ts_id:
- ts_id, version_date = ts_id.split(":", 1)
- version_date_dt = pd.to_datetime(version_date)
- else:
- version_date_dt = None
- data = get_timeseries(
- ts_id=ts_id,
- office_id=office_id,
- unit=unit,
- begin=begin,
- end=end,
- version_date=version_date_dt,
- multithread=False,
- )
- result_dict = {
- "ts_id": ts_id,
- "unit": data.json["units"],
- "version_date": version_date_dt,
- "values": data.df,
- }
- return result_dict
- except Exception as e:
- logging.error(f"Error processing {ts_id}: {e}")
- return None
+ if ":" in ts_id:
+ ts_id, version_date = ts_id.split(":", 1)
+ version_date_dt = pd.to_datetime(version_date)
+ else:
+ version_date_dt = None
+ data = get_timeseries(
+ ts_id=ts_id,
+ office_id=office_id,
+ unit=unit,
+ begin=begin,
+ end=end,
+ version_date=version_date_dt,
+ multithread=False,
+ )
+ result_dict = {
+ "ts_id": ts_id,
+ "unit": data.json["units"],
+ "version_date": version_date_dt,
+ "values": data.df,
+ }
+ return result_dict
+ logger.debug(
+ "Fetching %s time series with up to %s workers", len(ts_ids), max_workers
+ )
+ failures: list[tuple[str, Exception]] = []
+ result_dict = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
- results = executor.map(get_ts_ids, ts_ids)
+ futures = [(ts_id, executor.submit(get_ts_ids, ts_id)) for ts_id in ts_ids]
+ for ts_id, future in futures:
+ try:
+ result_dict.append(future.result())
+ except Exception as error:
+ failures.append((ts_id, error))
+ if failures:
+ raise api.BatchError(
+ f"{len(failures)} time series failed to fetch:", failures
+ ) from failures[0][1]
- result_dict = list(results)
data = pd.DataFrame()
for row in result_dict:
- if row:
- temp_df = row["values"]
- temp_df = temp_df.assign(ts_id=row["ts_id"], units=row["unit"])
- if "version_date" in row.keys():
- temp_df = temp_df.assign(version_date=row["version_date"])
- temp_df.dropna(how="all", axis=1, inplace=True)
- data = pd.concat([data, temp_df], ignore_index=True)
+ temp_df = row["values"]
+ temp_df = temp_df.assign(ts_id=row["ts_id"], units=row["unit"])
+ if "version_date" in row.keys():
+ temp_df = temp_df.assign(version_date=row["version_date"])
+ temp_df.dropna(how="all", axis=1, inplace=True)
+ data = pd.concat([data, temp_df], ignore_index=True)
if not melted and "date-time" in data.columns:
cols = ["ts_id", "units"]
@@ -234,16 +245,24 @@ def get_timeseries_chunk(
def _call_with_retry(fn: Any, *args: Any, attempts: int = _CHUNK_ATTEMPTS) -> Any:
+ if attempts < 1:
+ raise ValueError("attempts must be at least 1")
for i in range(attempts):
try:
return fn(*args)
- except Exception as e:
+ except (api.ApiError, ConnectionError, Timeout) as e:
status_code = getattr(getattr(e, "response", None), "status_code", None)
- if status_code == 404:
+ if isinstance(e, api.ApiError) and status_code not in {
+ 429,
+ 500,
+ 502,
+ 503,
+ 504,
+ }:
raise
if i == attempts - 1:
raise
- logging.warning(f"chunk attempt {i + 1}/{attempts} failed: {e}")
+ logger.warning(f"chunk attempt {i + 1}/{attempts} failed: {e}")
def fetch_timeseries_chunks(
@@ -254,7 +273,7 @@ def fetch_timeseries_chunks(
max_workers: int,
) -> List[Data]:
results: List[Data] = []
- errors: List[str] = []
+ errors: list[tuple[str, Exception]] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_chunk = {
@@ -278,14 +297,15 @@ def fetch_timeseries_chunks(
error_msg = (
f"Failed to fetch data from {chunk_start} to {chunk_end}: {e}"
)
- logging.error(error_msg)
- errors.append(error_msg)
+ logger.debug(error_msg)
+ errors.append(
+ (f"Failed to fetch data from {chunk_start} to {chunk_end}", e)
+ )
if errors:
- raise RuntimeError(
- f"{len(errors)} of {len(chunks)} chunk(s) failed to fetch:\n"
- + "\n".join(errors)
- )
+ raise api.BatchError(
+ f"{len(errors)} of {len(chunks)} chunk(s) failed to fetch:", errors
+ ) from errors[0][1]
return results
@@ -442,25 +462,14 @@ def get_timeseries(
# grab extents if begin is before CWMS DB were implemented to prevent empty queries outside of extents
if begin < datetime(2014, 1, 1, tzinfo=timezone.utc) and multithread:
- try:
- begin_extent, _, _ = get_ts_extents(ts_id=ts_id, office_id=office_id)
- # replace begin with begin extent if outside extents
- if begin < begin_extent:
- begin = begin_extent
- logging.debug(
- f"Requested begin was before any data in this timeseries. Reseting to {begin}"
- )
- except Exception as e:
- # If getting extents fails, fall back to single-threaded mode
- logging.debug(
- f"Could not retrieve time series extents ({e}). Falling back to single-threaded mode."
+ begin_extent, _, _ = get_ts_extents(ts_id=ts_id, office_id=office_id)
+ # replace begin with begin extent if outside extents
+ if begin < begin_extent:
+ begin = begin_extent
+ logger.debug(
+ f"Requested begin was before any data in this timeseries. Reseting to {begin}"
)
- response = api.get_with_paging(
- selector=selector, endpoint=endpoint, params=params
- )
- return Data(response, selector=selector)
-
chunk_size = (
timedelta(days=max_days_per_chunk)
if max_days_per_chunk is not None
@@ -478,7 +487,7 @@ def get_timeseries(
)
return Data(response, selector=selector)
else:
- logging.debug(
+ logger.debug(
f"Fetching {len(chunks)} chunks of timeseries data with {max_workers} threads"
)
# fetch the data
@@ -659,7 +668,7 @@ def store_ts_ids(
ts_data_all["ts_id"].astype(str) + ":" + ts_data_all["version_date"].astype(str)
).unique()
- errors: List[str] = []
+ errors: list[tuple[str, Exception]] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {}
for unique_tsid in unique_tsids:
@@ -685,12 +694,12 @@ def store_ts_ids(
try:
future.result()
except Exception as e:
- errors.append(f"{futures[future]}: {e}")
+ errors.append((str(futures[future]), e))
if errors:
- raise RuntimeError(
- f"{len(errors)} time series failed to store:\n" + "\n".join(errors)
- )
+ raise api.BatchError(
+ f"{len(errors)} time series failed to store:", errors
+ ) from errors[0][1]
def chunk_timeseries_data(
@@ -790,13 +799,13 @@ def store_timeseries(
_call_with_retry(api.post, endpoint, chunks[0], params)
remaining_chunks = chunks[1:]
actual_workers = min(max_workers, len(remaining_chunks))
- logging.debug(
+ logger.debug(
f"Storing {len(chunks)} chunks of timeseries data with {actual_workers} threads"
)
# Store chunks concurrently
responses: List[Dict[str, Any]] = []
- errors: List[str] = []
+ errors: list[tuple[str, Exception]] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=actual_workers) as executor:
future_to_chunk = {
@@ -812,15 +821,16 @@ def store_timeseries(
start_time = chunk["values"][0][0]
end_time = chunk["values"][-1][0]
error_msg = f"Error storing chunk from {start_time} to {end_time}: {e}"
- logging.error(error_msg)
- errors.append(error_msg)
+ logger.debug(error_msg)
+ errors.append(
+ (f"Error storing chunk from {start_time} to {end_time}", e)
+ )
responses.append({"error": error_msg})
if errors:
- raise RuntimeError(
- f"{len(errors)} of {len(chunks)} chunk(s) failed to store:\n"
- + "\n".join(errors)
- )
+ raise api.BatchError(
+ f"{len(errors)} of {len(chunks)} chunk(s) failed to store:", errors
+ ) from errors[0][1]
return
diff --git a/cwms/users/users.py b/cwms/users/users.py
index a69dc635..2119f606 100644
--- a/cwms/users/users.py
+++ b/cwms/users/users.py
@@ -1,4 +1,3 @@
-import json
from typing import Any, List, Optional
import cwms.api as api
@@ -14,7 +13,7 @@ def _raise_user_management_error(error: api.ApiError, action: str) -> None:
"are not authorized for user-management access or are missing the "
f"required role assignment. CDA responded with 403 {response_hint}."
)
- raise api.PermissionError(error.response, message) from None
+ raise api.PermissionError(error.response, message) from error
raise error
@@ -110,7 +109,7 @@ def get_user(user_name: str) -> dict[str, Any]:
if status_code == 404:
raise api.NotFoundError(
error.response, f"User '{user_name}' was not found."
- ) from None
+ ) from error
if status_code == 403:
_raise_user_management_error(error, f"User '{user_name}' retrieval")
raise
@@ -153,15 +152,10 @@ def delete_user_roles(user_name: str, office_id: str, roles: List[str]) -> None:
raise ValueError("Delete user roles requires a roles list")
endpoint = f"user/{user_name}/roles/{office_id}"
- headers = {"accept": "*/*", "Content-Type": api.api_version_text(api.API_VERSION)}
- # TODO: Delete does not currently support a body in the api module. Use SESSION directly
- with api.SESSION.delete(
- endpoint, headers=headers, data=json.dumps(roles)
- ) as response:
- if not response.ok:
- _raise_user_management_error(
- api.ApiError(response), f"User '{user_name}' role deletion"
- )
+ try:
+ api.delete(endpoint, data=roles)
+ except api.ApiError as error:
+ _raise_user_management_error(error, f"User '{user_name}' role deletion")
def update_user(user_name: str, office_id: str, roles: List[str]) -> None:
diff --git a/tests/cda/users/users_CDA_test.py b/tests/cda/users/users_CDA_test.py
index a311bea5..6f7d3ca4 100644
--- a/tests/cda/users/users_CDA_test.py
+++ b/tests/cda/users/users_CDA_test.py
@@ -87,11 +87,22 @@ def test_get_user_403_has_friendly_message_for_missing_roles(request):
with pytest.raises(cwms.PermissionError) as error:
cwms.get_user(user_name)
- assert str(error.value) == (
+ friendly_message = (
f"User '{user_name}' retrieval could not be completed because the current credentials "
"are not authorized for user-management access or are missing the required "
"role assignment. CDA responded with 403 Forbidden."
)
+ message = str(error.value)
+ response = error.value.response
+ assert friendly_message in message
+ assert response.status_code == 403
+ assert response.request.method == "GET"
+ assert response.url in message
+ assert "403 GET" in message
+ assert response.text
+ assert response.text in message
+ assert isinstance(error.value.__cause__, cwms.api.ApiError)
+ assert error.value.__cause__.response is response
def test_get_users():
diff --git a/tests/mock/api_error_test.py b/tests/mock/api_error_test.py
index 50f79683..a1fc18b6 100644
--- a/tests/mock/api_error_test.py
+++ b/tests/mock/api_error_test.py
@@ -1,6 +1,9 @@
from dataclasses import dataclass
from typing import Optional
+import pytest
+import requests
+
from cwms.api import ApiError
@@ -41,7 +44,7 @@ def test_api_error_str():
assert (
str(error)
- == "CWMS API Error (https://api.example.com/test) Not Found. May be the result of an empty query. incident identifier 34566432"
+ == "CWMS API Error (https://api.example.com/test) 404 Not Found. May be the result of an empty query. incident identifier 34566432"
)
# The response should not include a reason, since it is not included in the response.
@@ -50,11 +53,31 @@ def test_api_error_str():
assert (
str(error)
- == "CWMS API Error (https://api.example.com/test). May be the result of an empty query."
+ == "CWMS API Error (https://api.example.com/test) 404. May be the result of an empty query."
)
- # In the most minimal case, only the URL is included.
+ # Even without a reason or body, the URL and status are included.
response = Response(url="https://api.example.com/test", status_code=500)
error = ApiError(response)
- assert str(error) == "CWMS API Error (https://api.example.com/test)."
+ assert str(error) == "CWMS API Error (https://api.example.com/test) 500."
+
+
+@pytest.mark.parametrize("method", ["GET", "POST", "PATCH", "DELETE"])
+def test_404_hint_and_database_details_without_reason(method):
+ """Internal HTTP CDA writes can return 404 with a database incident body."""
+ response = requests.Response()
+ response.status_code = 404
+ response.url = "http://example.com/cwms-data/timeseries"
+ response.request = requests.Request(method, response.url).prepare()
+ response._content = (
+ b'{"message":"ORA-20998: ERROR",'
+ b'"incidentIdentifier":"test-incident","source":"Database","details":{}}'
+ )
+ error = ApiError(response)
+ message = str(error)
+ assert f"404 {method}" in message
+ assert response.url in message
+ assert response.text in message
+ assert ("empty query" in message) == (method == "GET")
+ assert error.response is response
diff --git a/tests/mock/api_test.py b/tests/mock/api_test.py
index 1b27d4e3..671888ab 100644
--- a/tests/mock/api_test.py
+++ b/tests/mock/api_test.py
@@ -8,6 +8,37 @@
TEST_ENDPOINT = "/test-endpoint"
+@pytest.mark.parametrize("scheme", ["http", "https"])
+def test_default_session_pool_for_both_protocols(scheme):
+ adapter = SESSION.get_adapter(f"{scheme}://example.com/cwms-data/")
+ pool = adapter.poolmanager.connection_from_url(f"{scheme}://example.com")
+ assert pool.pool.maxsize == 100
+ assert adapter.max_retries == cwms.api.retry_strategy
+
+
+@pytest.mark.parametrize("scheme", ["http", "https"])
+@pytest.mark.parametrize("pool_connections", [None, 37])
+def test_initialized_session_retains_concurrent_connections(
+ scheme, pool_connections, monkeypatch, caplog
+):
+ """Returning 30 workers' connections must not overflow Requests' default 10."""
+ monkeypatch.setattr(cwms.api, "SESSION", SESSION)
+ options = {} if pool_connections is None else {"pool_connections": pool_connections}
+ session = init_session(api_root=f"{scheme}://example.com/cwms-data", **options)
+ try:
+ adapter = session.get_adapter(session.base_url)
+ pool = adapter.poolmanager.connection_from_url(session.base_url)
+ assert pool.pool.maxsize == (pool_connections or 100)
+ assert adapter.max_retries == cwms.api.retry_strategy
+ # Exercise the real urllib3 return path without opening network sockets.
+ connections = [pool._get_conn() for _ in range(30)]
+ for connection in connections:
+ pool._put_conn(connection)
+ assert "Connection pool is full" not in caplog.text
+ finally:
+ session.close()
+
+
def test_session_default():
"""Verify the default root URL and auth headers."""
@@ -99,6 +130,23 @@ def post(self, endpoint, params=None, headers=None, data=None):
assert "incident identifier 34566432" in str(error.value)
+@pytest.mark.parametrize("method", ["get", "post", "patch", "delete"])
+def test_retry_error_retains_unknown_cause(monkeypatch, method):
+ original_error = ValueError("synthetic retry cause")
+ wrapped_error = RequestsRetryError(original_error)
+
+ def fail(*args, **kwargs):
+ raise wrapped_error
+
+ monkeypatch.setattr(cwms.api.SESSION, method, fail)
+ call = getattr(cwms.api, method)
+ kwargs = {"data": {}} if method in {"post", "patch"} else {}
+ with pytest.raises(RequestsRetryError) as caught:
+ call(TEST_ENDPOINT, **kwargs)
+ assert caught.value is wrapped_error
+ assert caught.value.args[0] is original_error
+
+
def test_retry_error_unwraps_original_cause(monkeypatch):
"""Verify wrapped retry failures propagate the underlying cause."""
diff --git a/tests/mock/error_handling_test.py b/tests/mock/error_handling_test.py
new file mode 100644
index 00000000..36cf8521
--- /dev/null
+++ b/tests/mock/error_handling_test.py
@@ -0,0 +1,274 @@
+"""Local error-path coverage; all HTTP responses are mocked."""
+
+import logging
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pandas as pd
+import pytest
+from requests import Response
+from requests.exceptions import Timeout
+
+import cwms.api as api
+import cwms.timeseries.timeseries as ts
+import cwms.users.users as users
+
+ROOT = "https://errors.example.test/"
+CALLS = [
+ ("GET", lambda: api.get("test")),
+ ("GET", lambda: api.get_xml("test")),
+ ("GET", lambda: api.get_with_paging("values", "test", {"page": None})),
+ ("POST", lambda: api.post("test", {})),
+ ("POST", lambda: api.post_with_returned_data("test", {})),
+ ("PATCH", lambda: api.patch("test", {})),
+ ("DELETE", lambda: api.delete("test")),
+]
+
+
+@pytest.fixture(autouse=True)
+def session(monkeypatch):
+ monkeypatch.setattr(api, "SESSION", api.sessions.BaseUrlSession(base_url=ROOT))
+
+
+@pytest.mark.parametrize("method,call", CALLS)
+@pytest.mark.parametrize("status", [400, 401, 403, 404, 409, 429, 500, 503])
+def test_http_errors_keep_response_details(requests_mock, method, call, status):
+ requests_mock.register_uri(
+ method,
+ ROOT + "test",
+ status_code=status,
+ json={"message": "synthetic failure", "incident-id": "test-123"},
+ )
+ with pytest.raises(api.ApiError) as caught:
+ call()
+ assert caught.value.response.status_code == status
+ assert str(status) in str(caught.value)
+ assert method in str(caught.value)
+ assert "synthetic failure" in str(caught.value)
+ assert "test-123" in str(caught.value)
+
+
+@pytest.mark.parametrize("method,call", CALLS)
+def test_transport_errors_propagate(requests_mock, method, call):
+ original = Timeout("synthetic timeout")
+ requests_mock.register_uri(method, ROOT + "test", exc=original)
+ with pytest.raises(Timeout) as caught:
+ call()
+ assert caught.value is original
+
+
+@pytest.mark.parametrize("method,call", [CALLS[0], CALLS[4]])
+def test_invalid_json_is_an_error(requests_mock, method, call):
+ requests_mock.register_uri(
+ method,
+ ROOT + "test",
+ text="{broken",
+ headers={"Content-Type": "application/json"},
+ )
+ with pytest.raises(api.ApiError, match="Invalid JSON") as caught:
+ call()
+ assert caught.value.response.text == "{broken"
+ assert isinstance(caught.value.__cause__, ValueError)
+
+
+@pytest.mark.parametrize(
+ "content_type,text,expected",
+ [
+ ("application/json", "", {}),
+ ("text/plain", "plain response", "plain response"),
+ ("application/xml", "", ""),
+ ],
+)
+def test_valid_response_formats(requests_mock, content_type, text, expected):
+ requests_mock.get(ROOT + "test", text=text, headers={"Content-Type": content_type})
+ assert api.get("test") == expected
+
+
+def test_later_page_error_is_not_partial_success(requests_mock):
+ requests_mock.get(
+ ROOT + "test",
+ [
+ {"json": {"values": [1], "next-page": "page-two"}},
+ {"status_code": 500, "text": "second page failed"},
+ ],
+ )
+ with pytest.raises(api.ApiError, match="second page failed"):
+ api.get_with_paging("values", "test", {"page": None})
+ assert requests_mock.call_count == 2
+
+
+def test_custom_user_error_keeps_server_details(requests_mock):
+ requests_mock.delete(
+ ROOT + "user/test-user/roles/TEST", status_code=403, text="incident test-456"
+ )
+ with pytest.raises(api.PermissionError) as caught:
+ users.delete_user_roles("test-user", "TEST", ["test-role"])
+ assert "incident test-456" in str(caught.value)
+ assert "role deletion" in str(caught.value)
+ assert isinstance(caught.value.__cause__, api.ApiError)
+ assert requests_mock.last_request.json() == ["test-role"]
+
+
+def test_debug_outcome_does_not_log_credentials_or_body(requests_mock, caplog):
+ api.SESSION.headers["Authorization"] = "Bearer synthetic-secret"
+ requests_mock.post(ROOT + "test", status_code=204)
+ with caplog.at_level(logging.DEBUG, logger="cwms.api"):
+ api.post("test", {"private": "synthetic-body"})
+ assert "CDA POST test returned HTTP 204" in caplog.text
+ assert "synthetic-secret" not in caplog.text
+ assert "synthetic-body" not in caplog.text
+
+
+def test_multi_fetch_reports_all_failures_without_partial_result(monkeypatch):
+ originals = {"bad-one": ValueError("first"), "bad-two": Timeout("second")}
+ visited = []
+
+ def fetch(ts_id, **kwargs):
+ visited.append(ts_id)
+ if ts_id in originals:
+ raise originals[ts_id]
+ return SimpleNamespace(json={"units": "ft"}, df=pd.DataFrame())
+
+ monkeypatch.setattr(ts, "get_timeseries", fetch)
+ with pytest.raises(RuntimeError) as caught:
+ ts.get_multi_timeseries_df(["bad-one", "good", "bad-two"], "TEST")
+ assert set(visited) == {"bad-one", "good", "bad-two"}
+ assert dict(caught.value.failures) == originals
+ assert caught.value.__cause__ in originals.values()
+
+
+def test_chunk_fetch_keeps_original_exceptions(monkeypatch):
+ original = ValueError("invalid chunk")
+ monkeypatch.setattr(ts, "get_timeseries_chunk", Mock(side_effect=original))
+ begin = datetime(2025, 1, 1, tzinfo=timezone.utc)
+ chunks = [
+ (begin, begin + timedelta(days=1)),
+ (begin + timedelta(days=1), begin + timedelta(days=2)),
+ ]
+ with pytest.raises(RuntimeError) as caught:
+ ts.fetch_timeseries_chunks(chunks, {}, "values", "timeseries", 2)
+ assert len(caught.value.failures) == 2
+ assert all(error is original for _, error in caught.value.failures)
+ assert caught.value.__cause__ is original
+
+
+@pytest.mark.parametrize("status", [400, 401, 403, 404, 409])
+def test_permanent_chunk_errors_are_not_retried(status):
+ response = Response()
+ response.status_code = status
+ original = api.ApiError(response)
+ call = Mock(side_effect=original)
+ with pytest.raises(api.ApiError) as caught:
+ ts._call_with_retry(call)
+ assert caught.value is original
+ assert call.call_count == 1
+
+
+def test_programming_errors_are_not_retried():
+ call = Mock(side_effect=TypeError("not serializable"))
+ with pytest.raises(TypeError):
+ ts._call_with_retry(call)
+ assert call.call_count == 1
+
+
+def test_transient_errors_retry_and_preserve_final_error():
+ original = Timeout("timeout")
+ call = Mock(side_effect=[original, "success"])
+ assert ts._call_with_retry(call, attempts=2) == "success"
+ call = Mock(side_effect=original)
+ with pytest.raises(Timeout) as caught:
+ ts._call_with_retry(call, attempts=2)
+ assert caught.value is original
+ assert call.call_count == 2
+
+
+def test_extent_failure_is_not_suppressed(monkeypatch):
+ original = ValueError("invalid extents")
+ monkeypatch.setattr(ts, "get_ts_extents", Mock(side_effect=original))
+ fallback = Mock()
+ monkeypatch.setattr(api, "get_with_paging", fallback)
+ begin = datetime(2010, 1, 1, tzinfo=timezone.utc)
+ with pytest.raises(ValueError, match="invalid extents"):
+ ts.get_timeseries("test-series", "TEST", begin=begin)
+ fallback.assert_not_called()
+
+
+def test_chunk_store_preserves_failures_and_initialization_order(monkeypatch):
+ original = ValueError("bad value")
+ visited = []
+
+ def post(endpoint, data, params):
+ value = data["values"][0][0]
+ if value != 0:
+ assert 0 in visited
+ visited.append(value)
+ if value == 1:
+ raise original
+
+ monkeypatch.setattr(api, "post", post)
+ data = {
+ "name": "test-series",
+ "office-id": "TEST",
+ "units": "ft",
+ "values": [[0, 1, 0], [1, 2, 0], [2, 3, 0]],
+ }
+ with pytest.raises(RuntimeError) as caught:
+ ts.store_timeseries(data, chunk_size=1)
+ assert sorted(visited) == [0, 1, 2]
+ assert caught.value.failures == [("Error storing chunk from 1 to 1", original)]
+ assert caught.value.__cause__ is original
+
+
+def test_initial_chunk_error_stops_remaining_writes(monkeypatch):
+ original = TypeError("invalid value")
+ post = Mock(side_effect=original)
+ monkeypatch.setattr(api, "post", post)
+ data = {
+ "name": "test-series",
+ "office-id": "TEST",
+ "units": "ft",
+ "values": [[0, 1, 0], [1, 2, 0]],
+ }
+ with pytest.raises(TypeError) as caught:
+ ts.store_timeseries(data, chunk_size=1)
+ assert caught.value is original
+ assert post.call_count == 1
+
+
+def test_multi_store_preserves_all_original_failures(monkeypatch):
+ originals = {"test-one": ValueError("first"), "test-two": TypeError("second")}
+
+ def store(data, **kwargs):
+ raise originals[data["name"]]
+
+ monkeypatch.setattr(ts, "store_timeseries", store)
+ data = pd.DataFrame(
+ {
+ "ts_id": list(originals),
+ "units": ["ft", "ft"],
+ "date-time": pd.to_datetime(["2025-01-01", "2025-01-01"], utc=True),
+ "value": [1.0, 2.0],
+ }
+ )
+ with pytest.raises(RuntimeError) as caught:
+ ts.store_multi_timeseries_df(data, "TEST")
+ assert {
+ context.split(":")[0]: error for context, error in caught.value.failures
+ } == originals
+
+
+@pytest.mark.parametrize("attempts", [0, -1])
+def test_invalid_retry_count_is_rejected(attempts):
+ call = Mock()
+ with pytest.raises(ValueError, match="attempts"):
+ ts._call_with_retry(call, attempts=attempts)
+
+
+@pytest.mark.parametrize("days", [0, -1])
+def test_invalid_chunk_duration_is_rejected(days):
+ begin = datetime(2025, 1, 1)
+ end = begin + timedelta(days=1)
+ chunk_size = timedelta(days=days)
+ with pytest.raises(ValueError, match="chunk_size"):
+ ts.chunk_timeseries_time_range(begin, end, chunk_size)
diff --git a/tests/mock/timeseries/timeseries_test.py b/tests/mock/timeseries/timeseries_test.py
index 911e77cc..8fe964e3 100644
--- a/tests/mock/timeseries/timeseries_test.py
+++ b/tests/mock/timeseries/timeseries_test.py
@@ -4,6 +4,7 @@
# Source may not be released without written approval from HEC
from datetime import datetime, timedelta
+from datetime import timezone as datetime_timezone
import pandas as pd
import pytest
@@ -140,7 +141,16 @@ def test_timeseries_df_to_json():
assert all(data == data2)
-def test_get_timeseries_unversioned_default(requests_mock):
+def test_get_timeseries_unversioned_default(requests_mock, monkeypatch):
+ monkeypatch.setattr(
+ timeseries,
+ "get_ts_extents",
+ lambda **kwargs: (
+ datetime(2008, 1, 1, tzinfo=datetime_timezone.utc),
+ None,
+ None,
+ ),
+ )
requests_mock.get(
f"{_MOCK_ROOT}"
"/timeseries?office=SWT&"
@@ -169,7 +179,16 @@ def test_get_timeseries_unversioned_default(requests_mock):
assert data.df.shape == (4, 3)
-def test_get_empty_ts_df(requests_mock):
+def test_get_empty_ts_df(requests_mock, monkeypatch):
+ monkeypatch.setattr(
+ timeseries,
+ "get_ts_extents",
+ lambda **kwargs: (
+ datetime(2008, 1, 1, tzinfo=datetime_timezone.utc),
+ None,
+ None,
+ ),
+ )
requests_mock.get(
f"{_MOCK_ROOT}"
"/timeseries?office=SWT&"
@@ -497,7 +516,16 @@ def test_create_timeseries_unversioned_default(requests_mock):
assert requests_mock.call_count == 1
-def test_get_timeseries_versioned_default(requests_mock):
+def test_get_timeseries_versioned_default(requests_mock, monkeypatch):
+ monkeypatch.setattr(
+ timeseries,
+ "get_ts_extents",
+ lambda **kwargs: (
+ datetime(2008, 1, 1, tzinfo=datetime_timezone.utc),
+ None,
+ None,
+ ),
+ )
requests_mock.get(
f"{_MOCK_ROOT}"
"/timeseries?office=SWT&"