krowvin · Sep 28, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026
diff --git 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
@@ -33,20 +33,22 @@
 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
 
 # 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 @@
         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 @@
         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 @@
         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 @@
     """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 @@
                 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 @@
         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 @@
             max_retries=retry_strategy,
         )
         SESSION.mount("https://", adapter)
+        SESSION.mount("http://", adapter)
     if token:
         if api_key:
             logging.warning(
@@ -336,6 +368,8 @@
 
 
 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 @@
         # 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 @@
     }
     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 @@
         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,24 +602,29 @@
         "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(
     endpoint: str,
     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 @@
     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 @@
         "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
@@ -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
@@ -127,7 +127,7 @@ def rating_spec_df_to_xml(data: pd.DataFrame) -> str:
     try:
         spec_xml += f"""
       <source-agency>{data.loc[0,'source-agency']}</source-agency>"""
-    except Exception:
+    except KeyError:
         spec_xml += """
       <source-agency/>"""
     spec_xml += f"""
@@ -154,7 +154,7 @@ def rating_spec_df_to_xml(data: pd.DataFrame) -> str:
     try:
         spec_xml2 += f"""
       <description>{data.loc[0,'description']}</description>"""
-    except Exception:
+    except KeyError:
         spec_xml2 += """
       <description/>"""
     spec_xml2 += """