Add bounded V3 request recovery and client sessions

.github/workflows/tests.yml

@@ -7,6 +7,8 @@ on:
   pull_request:
     types:
       - opened
+      - synchronize
+      - reopened
     branches:
       - 'master'
 
@@ -32,3 +34,7 @@ jobs:
           run: uv sync
         - name: Style
           run: uv run make lint
+        - name: Offline V3 tests
+          run: >-
+            uv run pytest tests/test_v3_authentication.py tests/test_v3_recovery.py
+            tests/test_v3_http.py tests/test_v3.py::test_validate_empty_response

README.rst

@@ -31,3 +31,55 @@ Raw data
         user=<user_id>, study_id=<self.study_id>
     )
 
+V3 request recovery
+===================
+
+Each V3 client owns its tokens, cookies, and connection pool. Use a context
+manager to release its HTTP connections after use:
+
+.. code-block:: python
+
+    from threading import Event
+    from actiapi.v3 import ActiGraphClientV3
+
+    cancelled = Event()
+    with ActiGraphClientV3(
+        api_access_key,
+        api_secret_key,
+        timeout=(3.05, 30.0),
+        max_retries=2,
+        max_retry_delay=30.0,
+        cancel_event=cancelled,
+    ) as client:
+        studies = client.get_studies()
+
+``timeout`` specifies connection and read inactivity limits in seconds, or a
+single positive value for both. GET requests can retry connection failures,
+timeouts, and HTTP 429, 500, 502, 503, and 504 responses. ``max_retries`` counts
+additional attempts; zero selects one attempt. Backoff starts at 0.5 seconds
+and doubles within ``max_retry_delay``.
+
+A valid ``Retry-After`` delay is honoured within that wait limit. A longer
+server delay returns the HTTP failure immediately so the caller can schedule
+a later attempt. Authentication POSTs use one attempt and surface redirects
+as an explicit error. A data request receiving
+HTTP 401 can refresh its scope token once and try again.
+
+Calling ``cancelled.set()`` stops a pending retry wait and prevents subsequent
+attempts. Cancellation is observed after an in-flight socket operation finishes
+or times out. Connection and read inactivity limits apply to that operation.
+
+Successful return shapes and existing 404 handling are preserved. HTTP failures
+raise ``requests.HTTPError``. Invalid JSON, missing authentication tokens, and
+malformed pages raise ``actiapi.v3.InvalidResponseError``. Cancellation raises
+``actiapi.v3.RequestCancelled``. Transport failures retain their Requests
+exception types. Each of these errors is a ``requests.RequestException``.
+
+The offline tests use synthetic responses and a loopback HTTP server:
+
+.. code-block:: bash
+
+    uv run pytest tests/test_v3_authentication.py tests/test_v3_recovery.py \
+        tests/test_v3_http.py tests/test_v3.py::test_validate_empty_response
+
+The other existing tests require access to the configured live study.

actiapi/_transport.py

@@ -0,0 +1,128 @@
+"""Bounded synchronous transport for the V3 example client."""
+
+import math
+import time
+from datetime import timezone
+from email.utils import parsedate_to_datetime
+from threading import Event
+from typing import Optional, Tuple, Union
+
+import requests
+
+
+class RequestCancelled(requests.RequestException):
+    """The caller cancelled a request or its retry wait."""
+
+
+class InvalidResponseError(requests.RequestException):
+    """The server response does not satisfy the expected JSON contract."""
+
+
+class Transport:
+    """Own one session with finite timeouts and bounded GET retries."""
+
+    RETRY_STATUSES = {429, 500, 502, 503, 504}
+
+    def __init__(
+        self,
+        timeout: Union[float, Tuple[float, float]],
+        max_retries: int,
+        max_retry_delay: float,
+        cancel_event: Optional[Event],
+    ):
+        """Validate policy before creating the client session."""
+        values = timeout if isinstance(timeout, tuple) else (timeout,)
+        if len(values) not in (1, 2) or any(
+            isinstance(value, bool)
+            or not isinstance(value, (int, float))
+            or not math.isfinite(value)
+            or value <= 0
+            for value in values
+        ):
+            raise ValueError("timeout must contain finite positive values")
+        if isinstance(timeout, tuple) and len(timeout) != 2:
+            raise ValueError("timeout tuple must contain connect and read values")
+        if type(max_retries) is not int or max_retries < 0:
+            raise ValueError("max_retries must be a nonnegative integer")
+        if (
+            isinstance(max_retry_delay, bool)
+            or not isinstance(max_retry_delay, (int, float))
+            or not math.isfinite(max_retry_delay)
+            or max_retry_delay < 0
+        ):
+            raise ValueError("max_retry_delay must be finite and nonnegative")
+        self.timeout = timeout
+        self.max_retries = max_retries
+        self.max_retry_delay = max_retry_delay
+        self.cancel_event = cancel_event
+        self.session = requests.Session()
+
+    def close(self):
+        """Release this client's connection pool."""
+        self.session.close()
+
+    def _check_cancelled(self):
+        if self.cancel_event is not None and self.cancel_event.is_set():
+            raise RequestCancelled("Request cancelled")
+
+    def _wait(self, delay):
+        if self.cancel_event is None:
+            time.sleep(delay)
+        elif self.cancel_event.wait(delay):
+            raise RequestCancelled("Request cancelled during retry wait")
+
+    def _retry_delay(self, response, fallback):
+        value = response.headers.get("Retry-After")
+        if value is None:
+            return fallback
+        try:
+            delay = int(value)
+            if delay < 0:
+                return fallback
+        except ValueError:
+            try:
+                date = parsedate_to_datetime(value)
+                if date.tzinfo is None:
+                    date = date.replace(tzinfo=timezone.utc)
+                delay = max(0.0, date.timestamp() - time.time())
+            except (ValueError, TypeError, OverflowError):
+                return fallback
+        if delay > self.max_retry_delay:
+            return None
+        return delay
+
+    def request(self, method, url, **kwargs):
+        """Retry transient GET failures within the configured attempt budget."""
+        retry_limit = self.max_retries if method.upper() == "GET" else 0
+        attempt = 0
+        backoff = 0.5
+        while True:
+            self._check_cancelled()
+            delay = min(backoff, self.max_retry_delay)
+            try:
+                response = self.session.request(
+                    method, url, timeout=self.timeout, **kwargs
+                )
+            except requests.exceptions.SSLError:
+                raise
+            except (requests.ConnectionError, requests.Timeout):
+                if attempt >= retry_limit:
+                    raise
+            else:
+                try:
+                    self._check_cancelled()
+                except RequestCancelled:
+                    response.close()
+                    raise
+                if (
+                    response.status_code not in self.RETRY_STATUSES
+                    or attempt >= retry_limit
+                ):
+                    return response
+                delay = self._retry_delay(response, delay)
+                if delay is None:
+                    return response
+                response.close()
+            self._wait(delay)
+            attempt += 1
+            backoff = min(backoff * 2, self.max_retry_delay)

actiapi/v3.py

@@ -4,15 +4,18 @@
 """
 
 import logging
-from collections import defaultdict
-from typing import Any, Dict, List, Literal, Optional, Union
-
-import requests
+from threading import Event
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union
 
 from actiapi import ActiGraphClient
+from actiapi._transport import InvalidResponseError, RequestCancelled, Transport
 
-tokens: defaultdict[str, Optional[str]] = defaultdict(lambda: None)
-session = requests.Session()
+__all__ = [
+    "ActiGraphClientV3",
+    "InvalidResponseError",
+    "RequestCancelled",
+    "validate_response",
+]
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +26,33 @@ class ActiGraphClientV3(ActiGraphClient):
     BASE_URL = "https://api.actigraphcorp.com"
     AUTH_API = "https://auth.actigraphcorp.com/connect/token"
 
+    def __init__(
+        self,
+        api_access_key: str,
+        api_secret_key: str,
+        *,
+        timeout: Union[float, Tuple[float, float]] = (3.05, 30.0),
+        max_retries: int = 2,
+        max_retry_delay: float = 30.0,
+        cancel_event: Optional[Event] = None,
+    ):
+        """Initialize independent token and transport state with bounded retries."""
+        super().__init__(api_access_key, api_secret_key)
+        self._tokens: Dict[str, Optional[str]] = {}
+        self._transport = Transport(timeout, max_retries, max_retry_delay, cancel_event)
+
+    def close(self):
+        """Release this client's HTTP connections."""
+        self._transport.close()
+
+    def __enter__(self):
+        """Enter the client context."""
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Release connections when leaving the client context."""
+        self.close()
+
     @staticmethod
     def _generate_headers(token: str, raw: bool = False):
         headers = {}
@@ -42,15 +72,28 @@ def _get_access_token(self, scope: str):
         }
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
 
-        response = requests.post(
-            endpoint, data=request_body, headers=headers, verify=True
-        )
-        try:
-            return response.json()["access_token"]
-        except KeyError:
-            raise RuntimeError(
-                "No access token! Make sure you have API_ACCESS_KEY and API_SECRET_KEY!"
-            )
+        with self._transport.request(
+            "POST",
+            endpoint,
+            data=request_body,
+            headers=headers,
+            verify=True,
+            allow_redirects=False,
+        ) as response:
+            if 300 <= response.status_code < 400:
+                raise InvalidResponseError(
+                    "Authentication endpoint returned a redirect"
+                )
+            reply = validate_response(response)
+            if (
+                not isinstance(reply, dict)
+                or not isinstance(reply.get("access_token"), str)
+                or not reply["access_token"]
+            ):
+                raise InvalidResponseError(
+                    "Authentication response has no access token"
+                )
+            return reply["access_token"]
 
     def get_files(
         self,
@@ -220,35 +263,39 @@ def get_sleep_summary(
         return results
 
     def _get_single(self, request: str, scope: str):
-        global tokens
-        if tokens[scope] is None:
-            tokens[scope] = self._get_access_token(scope)
-        logger.info("Requesting %s", request)
-        headers = self._generate_headers(str(tokens[scope]))
-        response = session.get(self.BASE_URL + request, headers=headers, stream=False)
-        reply = validate_response(response)
-        return reply
+        for refresh in range(2):
+            if self._tokens.get(scope) is None:
+                self._tokens[scope] = self._get_access_token(scope)
+            logger.info("Requesting %s", request)
+            headers = self._generate_headers(str(self._tokens[scope]))
+            with self._transport.request(
+                "GET", self.BASE_URL + request, headers=headers, stream=False
+            ) as response:
+                if response.status_code == 401:
+                    self._tokens.pop(scope, None)
+                    if refresh == 0:
+                        continue
+                return validate_response(response)
 
     def _get_paginated(self, request: str, scope: str):
-        global tokens
         results = []
         offset = 0
         limit = 100
         while True:
             paginated_request = f"{request}offset={offset}&limit={limit}"
-            try:
-                reply = self._get_single(request=paginated_request, scope=scope)
-                if reply is None:
-                    break
-
-                total_count = reply["totalCount"]
-            except KeyError:
-                tokens[scope] = None
-                reply = self._get_single(request=paginated_request, scope=scope)
-                if reply is None:
-                    break
-
-                total_count = reply["totalCount"]
+            reply = self._get_single(request=paginated_request, scope=scope)
+            if reply is None:
+                break
+            if (
+                not isinstance(reply, dict)
+                or type(reply.get("totalCount")) is not int
+                or reply["totalCount"] < 0
+                or not isinstance(reply.get("items"), list)
+            ):
+                raise InvalidResponseError("Page requires totalCount and items")
+            total_count = reply["totalCount"]
+            if not reply["items"] and offset < total_count:
+                raise InvalidResponseError("Empty page before totalCount was reached")
 
             for item in reply["items"]:
                 results.append(item)
@@ -267,5 +314,9 @@ def validate_response(response):
         logger.error("404 Not Found!")
         result = None
     else:
-        result = response.json()
+        response.raise_for_status()
+        try:
+            result = response.json()
+        except ValueError as error:
+            raise InvalidResponseError("API response is not valid JSON") from error
     return result

pyproject.toml

@@ -7,6 +7,7 @@ readme = "README.rst"
 license = "GPL-3.0-only"
 maintainers = [{ name = "ActiGraph Data Science Team", email = "science@theactigraph.com" }]
 dynamic = ["version"]
+dependencies = ["requests>=2.32.4,<3"]
 
 [project.urls]
 Repository = "https://github.com/actigraph/actiapi"