Merge branch 'main' into update/users

Author: Enovotny

CONTRIBUTING.md

@@ -51,6 +51,10 @@ poetry run pytest tests/turbines/turbines_test.py
 
 To test and develop with a local CDA instance, follow these steps:
 
+For application code and manual local development, `cwms.api.init_session()`
+now supports either `api_key=` or `token=` authentication. When both are
+passed, bearer token auth takes precedence.
+
 1. **Install Docker**  
     Download and install Docker from the [official website](https://www.docker.com/get-started/).
 
@@ -90,6 +94,10 @@ To test and develop with a local CDA instance, follow these steps:
     poetry run pytest tests/cda/ --api_root=http://localhost:8082/cwms-data/
     ```
 
+    At the moment, the CDA pytest options are still keyed to `--api_key`; the
+    new `token=` support is available through `cwms.api.init_session()` in
+    library code.
+
     > **Warning:** The tests in the `tests/cda/` package are destructive and may cause irreversible deletion of data from the targeted CDA instance. Use caution and avoid running these tests against production or important environments.
 
 ### Code Style
@@ -119,4 +127,4 @@ to grab the module from testpypi for testin run
 
 ```sh
 python3 -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple/ cwms-python
-```
+```

README.md

@@ -20,6 +20,24 @@ Then import the package:
 import cwms
 ```
 
+### Authentication
+
+`cwms.init_session()` supports both CDA API keys and Keycloak access tokens.
+Use `api_key=` for the headless CDA API key flow, or `token=` for an OIDC access
+token such as one saved by [`cwms-cli login`]().
+
+```python
+import cwms
+
+cwms.init_session(
+    api_root="https://cwms-data.usace.army.mil/cwms-data/",
+    token="ACCESS_TOKEN",
+)
+```
+
+If both `token` and `api_key` are provided, `cwms-python` will use the token
+and log a warning.
+
 ## Getting Started
 
 ```python

cwms/api.py

@@ -5,9 +5,9 @@
 interact with these directly.
 
 The `init_session()` function can be used to specify an alternative root URL, and to
-provide an authentication key (if required). If `init_session()` is not called, the
-default root URL (see `API_ROOT` below) will be used, and no authentication keys will be
-included when making API calls.
+provide an authentication key or bearer token (if required). If `init_session()` is not
+called, the default root URL (see `API_ROOT` below) will be used, and no authentication
+headers will be included when making API calls.
 
 Example: Initializing a session
 
@@ -17,6 +17,9 @@
     # Specify an alternate URL and an auth key
     init_session(api_root="https://example.com/cwms-data", api_key="API_KEY")
 
+    # Specify an alternate URL and an OIDC bearer token
+    init_session(api_root="https://example.com/cwms-data", token="ACCESS_TOKEN")
+
 Functions which make API calls that _may_ return a JSON response will return a `dict`
 containing the deserialized data. If the API response does not include data, an empty
 `dict` will be returned.
@@ -34,6 +37,7 @@
 from typing import Any, Optional, cast
 
 from requests import Response, adapters
+from requests.exceptions import RetryError as RequestsRetryError
 from requests_toolbelt import sessions  # type: ignore
 from requests_toolbelt.sessions import BaseUrlSession  # type: ignore
 from urllib3.util.retry import Retry
@@ -52,12 +56,12 @@
     status_forcelist=[
         403,
         429,
-        500,
         502,
         503,
         504,
     ],  # Example: also retry on these HTTP status codes
     allowed_methods=["GET", "PUT", "POST", "PATCH", "DELETE"],  # Methods to retry
+    raise_on_status=False,
 )
 SESSION = sessions.BaseUrlSession(base_url=API_ROOT)
 adapter = adapters.HTTPAdapter(
@@ -137,21 +141,46 @@ 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."""
+
+    current: Exception = error
+    cause = error.__cause__
+    while isinstance(cause, Exception):
+        current = cause
+        cause = cause.__cause__
+
+    if current is error and error.args:
+        first_arg = error.args[0]
+        if isinstance(first_arg, Exception):
+            current = first_arg
+            reason = getattr(current, "reason", None)
+            while isinstance(reason, Exception):
+                current = reason
+                reason = getattr(current, "reason", None)
+
+    return current
+
+
 def init_session(
     *,
     api_root: Optional[str] = None,
     api_key: Optional[str] = None,
+    token: Optional[str] = None,
     pool_connections: int = 100,
 ) -> BaseUrlSession:
-    """Specify a root URL and authentication key for the CWMS Data API.
+    """Specify a root URL and authentication credentials for the CWMS Data API.
 
     This function can be used to change the root URL used when interacting with the CDA.
-    All API calls made after this function is called will use the specified URL. If an
-    authentication key is given it will be included in all future request headers.
+    All API calls made after this function is called will use the specified URL. If
+    authentication credentials are given they will be included in all future request
+    headers.
 
     Keyword Args:
         api_root (optional): The root URL for the CWMS Data API.
         api_key (optional): An authentication key.
+        token (optional): A Keycloak access token. If both token and api_key are
+            provided, token is used.
 
     Returns:
         Returns the updated session object.
@@ -169,7 +198,16 @@ def init_session(
             max_retries=retry_strategy,
         )
         SESSION.mount("https://", adapter)
-    if api_key:
+    if token:
+        if api_key:
+            logging.warning(
+                "Both token and api_key were provided to init_session(); using token for Authorization."
+            )
+        # Ensure we don't provide the bearer text twice
+        if token.lower().startswith("bearer "):
+            token = token[7:]
+        SESSION.headers.update({"Authorization": "Bearer " + token})
+    elif api_key:
         if api_key.startswith("apikey "):
             api_key = api_key.replace("apikey ", "")
         SESSION.headers.update({"Authorization": "apikey " + api_key})
@@ -292,11 +330,14 @@ def get(
     """
 
     headers = {"Accept": api_version_text(api_version)}
-    with SESSION.get(endpoint, params=params, headers=headers) as response:
-        if not response.ok:
-            logging.error(f"CDA Error: response={response}")
-            raise ApiError(response)
-        return _process_response(response)
+    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)
+            return _process_response(response)
+    except RequestsRetryError as error:
+        raise _unwrap_retry_error(error) from None
 
 
 def get_with_paging(
@@ -351,11 +392,16 @@ def _post_function(
     headers = {"accept": "*/*", "Content-Type": api_version_text(api_version)}
     if isinstance(data, dict) or isinstance(data, list):
         data = json.dumps(data)
-    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)
-        return response
+    try:
+        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)
+            return response
+    except RequestsRetryError as error:
+        raise _unwrap_retry_error(error) from None
 
 
 def post(
@@ -445,10 +491,15 @@ def patch(
 
     if data and isinstance(data, dict) or isinstance(data, list):
         data = json.dumps(data)
-    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)
+    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)
+    except RequestsRetryError as error:
+        raise _unwrap_retry_error(error) from None
 
 
 def delete(
@@ -472,7 +523,10 @@ def delete(
     """
 
     headers = {"Accept": api_version_text(api_version)}
-    with SESSION.delete(endpoint, params=params, headers=headers) as response:
-        if not response.ok:
-            logging.error(f"CDA Error: response={response}")
-            raise ApiError(response)
+    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)
+    except RequestsRetryError as error:
+        raise _unwrap_retry_error(error) from None

examples/other/cwms_token_profile_check.py

@@ -0,0 +1,49 @@
+import argparse
+import json
+import os
+from getpass import getpass
+
+import cwms
+from cwms.api import ApiError
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Check CWMS token auth by fetching the authenticated user profile."
+    )
+    parser.add_argument(
+        "--api-root",
+        default=os.getenv("CDA_API_ROOT"),
+        help="CWMS Data API root URL.",
+    )
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    if not args.api_root:
+        print(
+            "Error: API root URL is required. Provide it with --api-root or set CDA_API_ROOT environment variable."
+        )
+        raise SystemExit(1)
+    token = getpass("Paste Keycloak access token: ").strip()
+    if not token:
+        raise SystemExit("No token provided.")
+
+    cwms.init_session(api_root=args.api_root, token=token)
+
+    try:
+        profile = cwms.api.get("auth/profile", api_version=1)
+        endpoint = "auth/profile"
+    except ApiError as error:
+        if getattr(error.response, "status_code", None) != 404:
+            raise
+        profile = cwms.get_user_profile()
+        endpoint = "user/profile"
+
+    print(f"Fetched profile from {endpoint}:")
+    print(json.dumps(profile, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

tests/mock/api_test.py

@@ -1,6 +1,11 @@
 import pytest
+from requests.exceptions import RetryError as RequestsRetryError
+from urllib3.exceptions import MaxRetryError, ResponseError
 
-from cwms.api import SESSION, InvalidVersion, api_version_text, init_session
+import cwms.api
+from cwms.api import SESSION, ApiError, api_version_text, init_session
+
+TEST_ENDPOINT = "/test-endpoint"
 
 
 def test_session_default():
@@ -53,3 +58,60 @@ def test_api_headers():
 
     version = api_version_text(api_version=2)
     assert version == "application/json;version=2"
+
+
+def test_retry_strategy_configuration():
+    """Verify retry behavior preserves the original CDA error path."""
+
+    retries = SESSION.adapters["https://"].max_retries
+
+    assert 500 not in retries.status_forcelist
+    assert retries.raise_on_status is False
+
+
+def test_post_500_raises_api_error(monkeypatch):
+    """Verify a 500 response is surfaced directly as ApiError."""
+
+    class ResponseStub:
+        url = "https://example.com/cwms-data/test-endpoint"
+        ok = False
+        status_code = 500
+        reason = "Internal Server Error"
+        content = b"incident identifier 34566432"
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    class SessionStub:
+        def post(self, endpoint, params=None, headers=None, data=None):
+            return ResponseStub()
+
+    monkeypatch.setattr(cwms.api, "SESSION", SessionStub())
+
+    with pytest.raises(ApiError) as error:
+        cwms.api._post_function(endpoint=TEST_ENDPOINT, data={})
+
+    assert error.value.response.status_code == 500
+    assert "Internal Server Error" in str(error.value)
+    assert "incident identifier 34566432" in str(error.value)
+
+
+def test_retry_error_unwraps_original_cause(monkeypatch):
+    """Verify wrapped retry failures propagate the underlying cause."""
+
+    original_error = ResponseError("too many 503 error responses")
+    wrapped_error = RequestsRetryError(
+        MaxRetryError(pool=None, url=TEST_ENDPOINT, reason=original_error)
+    )
+
+    class SessionStub:
+        def get(self, endpoint, params=None, headers=None):
+            raise wrapped_error
+
+    monkeypatch.setattr(cwms.api, "SESSION", SessionStub())
+
+    with pytest.raises(ResponseError, match="too many 503 error responses"):
+        cwms.api.get(TEST_ENDPOINT)