Add _BaseClient class to waterdata_client package (#304)

* enable direct injection of url request builder callable

* generic client module

* add dev notes to class docstring

* add attribute checking for subclasses

* add minimal testing for generic client"

* add high level get method requirement

* rename generic client

* setup request builder on init

* move URL building to separate method

* make get responses more specific

* add tests for max pages and other value checks

* add tests for bad responses

* use more accurate definition for QueryType

* add tests for different query types

* add pagination handling tests

* remove _ private class and module designation

Author: jarq6c

python/waterdata_client/src/hydrotools/waterdata_client/async_web_client.py

@@ -39,6 +39,7 @@
 from typing import Any, Optional, Self, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from enum import StrEnum
+from json import JSONDecodeError
 
 import aiohttp
 from tenacity import (
@@ -172,7 +173,19 @@ async def _execute_request(
                     return None
 
                 if content_type == ResponseContentType.JSON:
-                    return await response.json()
+                    try:
+                        return await response.json()
+                    except (
+                        aiohttp.ContentTypeError,
+                        ValueError,
+                        JSONDecodeError
+                    ) as exc:
+                        LOGGER.error(
+                            "Failed to decode JSON from %s: %s",
+                            url,
+                            exc
+                        )
+                        return None
                 return await response.read()
 
     async def fetch_all(

python/waterdata_client/src/hydrotools/waterdata_client/base_client.py

@@ -0,0 +1,176 @@
+"""Base OGC API Client.
+
+This module provides the base class for all collection-specific USGS OGC API 
+clients. It handles configuration state and the low-level request pipeline.
+"""
+import ssl
+from typing import Any, Optional, Sequence
+from functools import partial
+
+from yarl import URL
+
+from .async_web_client import get_all, ResponseContentType
+from .client_config import SETTINGS
+from .constants import USGSCollection, OGCPATH, OGCAPI
+from .url_builder import (
+    build_request,
+    build_request_batch,
+    build_request_batch_from_queries,
+    build_request_batch_from_feature_ids,
+    QueryType
+)
+
+class BaseClient:
+    """Base class for USGS OGC API clients. Specific child classes may overwrite
+    private attributes: _server, _api, _endpoint, _path, _content_type,
+    _max_pages.
+    Otherwise, these are set to package SETTINGS defaults.
+
+    Attributes:
+        concurrency_limit: Max simultaneous requests allowed for this client.
+        max_retries: Number of times to attempt a failed request.
+        timeout_seconds: Total request timeout in seconds.
+        ssl_context: Custom SSL context for requests.
+    """
+    _server: URL = SETTINGS.usgs_base_url
+    _api: OGCAPI = SETTINGS.default_api
+    _endpoint: USGSCollection = SETTINGS.default_collection
+    _path: OGCPATH = SETTINGS.default_path
+    _content_type: ResponseContentType = ResponseContentType.JSON
+    _max_pages: int = SETTINGS.max_pages
+
+    def __init__(
+        self,
+        concurrency_limit: int = SETTINGS.default_concurrency,
+        max_retries: int = SETTINGS.default_retries,
+        timeout_seconds: int = SETTINGS.timeout_seconds,
+        ssl_context: Optional[ssl.SSLContext] = None
+    ) -> None:
+        self.concurrency_limit = concurrency_limit
+        self.max_retries = max_retries
+        self.timeout_seconds = timeout_seconds
+        self.ssl_context = ssl_context
+
+        # Setup request builder
+        self._builder = partial(
+            build_request,
+            server=self._server,
+            api=self._api,
+            endpoint=self._endpoint,
+            path=self._path
+            )
+
+    def __init_subclass__(cls):
+        super().__init_subclass__()
+
+        # Enfore required attributes
+        required = ["_endpoint", "_path", "_api", "_server", "_content_type"]
+        for attr in required:
+            if not hasattr(cls, attr) or getattr(cls, attr) is None:
+                raise TypeError(
+                    f"Class {cls.__name__} failed to define required attribute: {attr}"
+                )
+
+        # Check for `get` method
+        if not callable(getattr(cls, "get", None)):
+            raise NotImplementedError(
+                f"Class {cls.__name__} must implement a public 'get' method "
+                "that wraps the internal '_get_responses' pipeline."
+            )
+
+    def _build_urls(
+        self,
+        feature_ids: Optional[Sequence[str]] = None,
+        queries: Optional[Sequence[QueryType]] = None
+    ) -> list[URL]:
+        """Constructs a list of yarl.URL objects given arguments.
+
+        Args:
+            feature_ids: Sequence of specific feature identifiers.
+            queries: A sequence of query parameter dictionaries.
+
+        Returns:
+            A list of yarl.URL objects.
+
+        """
+        if (feature_ids is not None) and (queries is not None):
+            return build_request_batch(
+                feature_ids=feature_ids,
+                queries=queries,
+                request_builder=self._builder
+            )
+        elif queries is not None:
+            return build_request_batch_from_queries(
+                queries=queries,
+                request_builder=self._builder
+                )
+        elif feature_ids is not None:
+            return build_request_batch_from_feature_ids(
+                feature_ids=feature_ids,
+                request_builder=self._builder
+                )
+        return [self._builder()]
+
+    def _get_json_responses(
+        self,
+        feature_ids: Optional[Sequence[str]] = None,
+        queries: Optional[Sequence[QueryType]] = None
+    ) -> list[dict[str, Any]]:
+        """Internal method to build URLs and execute concurrent requests. Note
+        that this method will silently drop bytes and None responses returned
+        by `get_all`. These responses are logged in `async_web_client`.
+
+        Args:
+            feature_ids: Sequence of specific feature identifiers.
+            queries: A sequence of query parameter dictionaries.
+
+        Returns:
+            A list of responses. Paginated responses are appended to the end
+            of the list.
+
+        """
+        # Prepare initial fetch
+        urls = self._build_urls(feature_ids, queries)
+        results: list[dict[str, Any]] = []
+
+        # Fetch data
+        for _ in range(self._max_pages):
+            # Get batch of URLs
+            batch = get_all(
+                urls=urls,
+                concurrency_limit=self.concurrency_limit,
+                max_retries=self.max_retries,
+                ssl_context=self.ssl_context,
+                timeout_seconds=self.timeout_seconds,
+                content_type=self._content_type
+            )
+
+            # Filter batch
+            json_batch: list[dict[str, Any]] = [b for b in batch if isinstance(b, dict)]
+
+            # Extend results
+            results.extend(json_batch)
+
+            # Inspect links for pagination
+            urls = [self._get_next_url(r) for r in json_batch if self._has_next_url(r)]
+
+            # No more data to fetch
+            if not urls:
+                break
+        return results
+
+    def _has_next_url(self, response: dict[str, Any] | bytes | None) -> bool:
+        """Checks response for the presence of pagination 'next' link."""
+        if not isinstance(response, dict):
+            return False
+        return any(link.get("rel") == "next" for link in response.get("links", []))
+
+    def _get_next_url(self, response: dict[str, Any]) -> URL:
+        """Attempts to return the first 'next' link encountered in response."""
+        if not isinstance(response, dict):
+            raise TypeError("response is not a dict")
+
+        for link in response.get("links", []):
+            if link.get("rel") == "next":
+                return URL(link["href"])
+        raise KeyError("response does not contain 'next' link")

python/waterdata_client/src/hydrotools/waterdata_client/client_config.py

@@ -72,6 +72,7 @@ class EnvironmentKey(StrEnum):
     RETRIES = f"{_KEY_START}RETRIES"
     TIMEOUT = f"{_KEY_START}TIMEOUT"
     API_KEY = f"{_KEY_START}USGS_API_KEY"
+    MAX_PAGES = f"{_KEY_START}MAX_PAGES"
 
     @classmethod
     def describe_keys(cls) -> str:
@@ -112,6 +113,8 @@ class _Settings:
             default_query.
         default_cache: Returns a diskcache.Cache object parameterized by cache_dir and
             cache_expires.
+        max_pages: Maximum number of times to follow paginated 'next' links
+            in JSON responses.
     """
     usgs_base_url: URL = URL("https://api.waterdata.usgs.gov/ogcapi/v0")
     schema_path: str = "openapi"
@@ -125,6 +128,7 @@ class _Settings:
     default_retries: int = 3
     timeout_seconds: int = 900
     usgs_api_key: Optional[str] = None
+    max_pages: int = 20
 
     @classmethod
     def from_env(cls) -> Self:
@@ -149,7 +153,8 @@ def from_env(cls) -> Self:
             default_concurrency=int(os.getenv(EnvironmentKey.CONCURRENCY, cls.default_concurrency)),
             default_retries=int(os.getenv(EnvironmentKey.RETRIES, cls.default_retries)),
             timeout_seconds=int(os.getenv(EnvironmentKey.TIMEOUT, cls.timeout_seconds)),
-            usgs_api_key=os.getenv(EnvironmentKey.API_KEY, cls.usgs_api_key)
+            usgs_api_key=os.getenv(EnvironmentKey.API_KEY, cls.usgs_api_key),
+            max_pages=int(os.getenv(EnvironmentKey.MAX_PAGES, cls.max_pages))
             )
 
     @cached_property

python/waterdata_client/src/hydrotools/waterdata_client/url_builder.py

@@ -1,33 +1,50 @@
 """USGS OGC API URL builders."""
-from typing import Optional, Sequence, Any
+from typing import Optional, Sequence, Any, Protocol
 
 from yarl import URL
 from multidict import MultiDict
 
 from .client_config import SETTINGS
 from .constants import OGCAPI, OGCPATH, USGSCollection
 
-QueryType = dict[str, Any] | MultiDict[str] | Sequence[tuple[str, Any]]
+QueryType = dict[str, Any] | MultiDict[Any] | Sequence[tuple[str, Any]]
 """Type alias for USGS OGC API compatible queryables."""
 
+class RequestBuilder(Protocol):
+    """Defines a Protocol for a callable object that builds USGS OGC compliant
+    URLS.
+    
+    Args:
+        feature_id: Optional specific feature identifier.
+        query: Optional query parameters.
+    
+    Returns:
+        A yarl.URL object.
+    """
+    def __call__(
+            self,
+            feature_id: Optional[str] = None,
+            query: Optional[QueryType] = None
+        ) -> URL: ...
+
 def build_request(
+    feature_id: Optional[str] = None,
+    query: Optional[QueryType] = None,
     server: URL = SETTINGS.usgs_base_url,
     api: OGCAPI = SETTINGS.default_api,
     endpoint: USGSCollection = SETTINGS.default_collection,
-    path: OGCPATH = SETTINGS.default_path,
-    feature_id: Optional[str] = None,
-    query: Optional[QueryType] = None
+    path: OGCPATH = SETTINGS.default_path
 ) -> URL:
     """Constructs a single yarl.URL.
 
     Args:
+        feature_id: Optional specific feature identifier.
+        query: Optional query parameters.
         server: The root URL for USGS OGC API services.
             URL('https://api.waterdata.usgs.gov/ogcapi/v0')
         api: USGS OGC API (e.g. 'collections').
         endpoint: USGS OGC API collection (e.g. 'continuous').
         path: USGS OGC path (e.g. 'items')
-        feature_id: Optional specific feature identifier.
-        query: Optional query parameters.
 
     Returns:
         A yarl.URL object.
@@ -47,13 +64,15 @@ def build_request(
 
 def build_request_batch(
     feature_ids: Sequence[str],
-    queries: Sequence[QueryType]
+    queries: Sequence[QueryType],
+    request_builder: RequestBuilder = build_request
 ) -> list[URL]:
     """Constructs a list of yarl.URL objects for paired IDs and queries.
 
     Args:
         feature_ids: Sequence of specific feature identifiers.
         queries: Sequence of query parameters corresponding to each ID.
+        request_builder: A callable that builds URLs.
 
     Returns:
         A list of yarl.URL objects.
@@ -66,30 +85,34 @@ def build_request_batch(
             f"Mismatched input lengths: feature_ids has {len(feature_ids)} items, "
             f"but queries has {len(queries)} items. Sequences must be of equal length."
         )
-    return [build_request(feature_id=f, query=q) for f, q in zip(feature_ids, queries)]
+    return [request_builder(feature_id=f, query=q) for f, q in zip(feature_ids, queries)]
 
 def build_request_batch_from_feature_ids(
-    feature_ids: Sequence[str]
+    feature_ids: Sequence[str],
+    request_builder: RequestBuilder = build_request
 ) -> list[URL]:
     """Constructs a list of yarl.URL objects for a sequence of feature IDs.
 
     Args:
         feature_ids: Sequence of specific feature identifiers.
+        request_builder: A callable that builds URLs.
 
     Returns:
         A list of yarl.URL objects.
     """
-    return [build_request(feature_id=f) for f in feature_ids]
+    return [request_builder(feature_id=f) for f in feature_ids]
 
 def build_request_batch_from_queries(
-    queries: Sequence[QueryType]
+    queries: Sequence[QueryType],
+    request_builder: RequestBuilder = build_request
 ) -> list[URL]:
     """Constructs a list of yarl.URL objects for a sequence of query dictionaries.
 
     Args:
         queries: Sequence of query parameters to add or override default_query.
+        request_builder: A callable that builds URLs.
 
     Returns:
         A list of yarl.URL objects.
     """
-    return [build_request(query=q) for q in queries]
+    return [request_builder(query=q) for q in queries]

python/waterdata_client/tests/test_async_web_client.py

@@ -45,11 +45,17 @@ async def error_handler(request: web.Request) -> web.Response:
             return web.Response(status=400)
         return web.Response(status=500)
 
+    async def malformed_json_handler(request: web.Request) -> web.Response:
+        if request.path != "/binary":
+            return web.Response(status=400)
+        return web.Response(body="{'invalid': json", content_type="application/json")
+
     app = web.Application()
     app.router.add_get("/json", json_handler)
     app.router.add_get("/json2", json2_handler)
     app.router.add_get("/binary", binary_handler)
     app.router.add_get("/error", error_handler)
+    app.router.add_get("/malformed", malformed_json_handler)
     return app
 
 async def test_fetch_json(aiohttp_client):
@@ -169,3 +175,21 @@ def test_get_all_bytes(persistent_server) -> None:
 
     assert results[0] == b"binary_data"
     assert results[1] == b"binary_data"
+
+async def test_fetch_invalid_json(aiohttp_client):
+    """Verifies that malformed JSON returns None rather than raising an error."""
+    mock_client = await aiohttp_client(create_web_application())
+    url = URL(f"http://{mock_client.host}:{mock_client.port}/malformed")
+
+    async with AsyncWebClient() as client:
+        result = await client.fetch(url)
+    assert result is None
+
+async def test_fetch_status_404_returns_none(aiohttp_client):
+    """Verifies that 4xx errors (non-retriable) return None immediately."""
+    mock_client = await aiohttp_client(create_web_application())
+    url = URL(f"http://{mock_client.host}:{mock_client.port}/missing_path")
+
+    async with AsyncWebClient() as client:
+        result = await client.fetch(url)
+    assert result is None