feat: add shared retrying requests session client (1/5)

CHANGELOG.md

@@ -8,7 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 (post 4.2.0 release)
 ### Added
 - Support for new OGC endpoints
+- Shared retrying `requests.Session` client in `maap.utils.requests_utils` (retries on transient errors for idempotent methods, default timeouts, request-time auth, token redaction in logs) ([#137](https://github.com/MAAP-Project/maap-py/issues/137))
 ### Changed
+- `requests_utils.make_request`/`make_dps_request` now retry transient failures, apply a default timeout and support PUT/DELETE
 ### Deprecated
 ### Removed
 - Removed functions calling old WPST endpoints

maap/utils/requests_utils.py

@@ -1,8 +1,12 @@
 import os
+import threading
+from typing import Optional, Union
 from maap.config_reader import MaapConfig
 import logging
 import requests
 from enum import Enum
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +25,65 @@
 DELETE = HTTPMethod.DELETE
 
 
+# Default (connect, read) timeout in seconds applied to every request unless overridden.
+DEFAULT_TIMEOUT = (10, 60)
+
+RETRY_STATUS_CODES = (429, 500, 502, 503, 504)
+IDEMPOTENT_METHODS = frozenset({'GET', 'HEAD', 'OPTIONS', 'PUT', 'DELETE'})
+RETRY_TOTAL = 3
+RETRY_BACKOFF_FACTOR = 0.5
+
+_REDACTED_HEADERS = {'token', 'authorization', 'proxy-ticket'}
+
+_sessions = {}
+_sessions_lock = threading.Lock()
+
+
+def _build_retry(allowed_methods) -> Retry:
+    return Retry(
+        total=RETRY_TOTAL,
+        backoff_factor=RETRY_BACKOFF_FACTOR,
+        status_forcelist=RETRY_STATUS_CODES,
+        allowed_methods=allowed_methods,
+        respect_retry_after_header=True,
+        # Hand the last response back to the caller instead of raising RetryError
+        raise_on_status=False,
+    )
+
+
+def _build_session(retry_post: bool) -> requests.Session:
+    methods = IDEMPOTENT_METHODS | {'POST'} if retry_post else IDEMPOTENT_METHODS
+    adapter = HTTPAdapter(max_retries=_build_retry(methods))
+    session = requests.Session()
+    session.mount('https://', adapter)
+    session.mount('http://', adapter)
+    return session
+
+
+def get_session(retry_post: bool = False) -> requests.Session:
+    """
+    Return the shared, lazily created ``requests.Session`` used by maap-py.
+
+    The session pools connections and retries transient failures (connection
+    errors and HTTP 429/500/502/503/504) with exponential backoff, honouring
+    ``Retry-After``. Only idempotent methods (GET, HEAD, OPTIONS, PUT, DELETE)
+    are retried. A second session that also retries POST is available with
+    ``retry_post=True`` for endpoints known to be safe to repeat.
+    """
+    with _sessions_lock:
+        if retry_post not in _sessions:
+            _sessions[retry_post] = _build_session(retry_post)
+        return _sessions[retry_post]
+
+
+def reset_sessions():
+    """Close and discard the shared sessions (mainly useful for tests)."""
+    with _sessions_lock:
+        for session in _sessions.values():
+            session.close()
+        _sessions.clear()
+
+
 def generate_dps_headers(config: MaapConfig, content_type=None):
     api_header = {
         'Accept': config.content_type,
@@ -37,31 +100,73 @@
     return api_header
 
 
+def _redact(headers: dict) -> dict:
+    return {k: ('***' if k.lower() in _REDACTED_HEADERS else v) for k, v in headers.items()}
+
+
 def check_response(dps_response):
     # if dps_response.status_code not in [200, 201]:
     #     raise RuntimeError('response is not 200 or 201. code: {}. details: {}'.format(dps_response.status_code,
     #                                                                                   dps_response.content))
     return dps_response.content.decode('UTF-8')
 
 
-# TODO: Explore consolidating all requests from maap-py into this class
+def request(method: Union[HTTPMethod, str], url, config: Optional[MaapConfig] = None, content_type=None,
+            auth: bool = True, headers: Optional[dict] = None, timeout=DEFAULT_TIMEOUT,
+            retry_post: bool = False, self_signed: bool = False, **kwargs) -> requests.Response:
+    """
+    Send an HTTP request through the shared retrying session.
+
+    Parameters
+    ----------
+    method : HTTPMethod or str
+        HTTP method.
+    url : str
+        Request URL.
+    config : MaapConfig, optional
+        Source of the token and default content type. Required when ``auth`` is True.
+        The token is read on every call, so a refreshed token is picked up automatically.
+    content_type : str, optional
+        Content-Type header to send.
+    auth : bool
+        If True, MAAP auth headers (token/Authorization, MAAP_PGT) are added. Set to False
+        for third-party hosts (e.g. DAAC downloads) so MAAP credentials are never sent.
+    headers : dict, optional
+        Extra headers; these take precedence over the generated ones. Never mutated.
+    timeout : float or tuple
+        ``requests`` timeout; defaults to ``DEFAULT_TIMEOUT``. Pass ``None`` to disable.
+    retry_post : bool
+        Also retry POST requests. Off by default as POSTs (e.g. job submission) may not be idempotent.
+    self_signed : bool
+        Disable TLS verification.
+    **kwargs
+        Passed through to ``requests.Session.request`` (``params``, ``json``, ``data``, ``stream``...).
+    """
+    method_name = method.value if isinstance(method, HTTPMethod) else str(method).upper()
+    if auth:
+        if config is None:
+            raise ValueError("config is required when auth=True")
+        merged = generate_dps_headers(config, content_type)
+    else:
+        merged = {'Content-Type': content_type} if content_type else {}
+    if headers:
+        merged.update(headers)
+
+    logger.debug(f"{method_name} request sent to {url}")
+    logger.debug(f"headers: {_redact(merged)}")
+    return get_session(retry_post).request(
+        method=method_name,
+        url=url,
+        headers=merged,
+        timeout=timeout,
+        verify=not self_signed,
+        **kwargs
+    )
+
+
 def make_request(url, config: MaapConfig, content_type=None, request_type: HTTPMethod = HTTPMethod.GET,
                  self_signed=False, **kwargs):
-    headers = generate_dps_headers(config, content_type)
-    logger.debug(f"{request_type} request sent to {url}")
-    logger.debug('headers:')
-    logger.debug(headers)
-    if request_type not in {POST, GET}:
-        # TODO: Add support for request type DELETE
-        raise NotImplementedError(f"Request type {request_type} not supported")
-    else:
-        return requests.request(
-            method=request_type.value,
-            url=url,
-            verify=not self_signed,
-            headers=headers,
-            ** kwargs
-        )
+    return request(request_type, url, config, content_type=content_type, self_signed=self_signed, **kwargs)
 
 
 def make_dps_request(url, config: MaapConfig, content_type=None, request_type: HTTPMethod = HTTPMethod.GET,

test/test_requests_utils.py

@@ -0,0 +1,172 @@
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from types import SimpleNamespace
+
+import pytest
+import requests
+
+from maap.utils import requests_utils as ru
+
+
+@pytest.fixture(autouse=True)
+def fresh_sessions(monkeypatch):
+    # Keep tests fast: no real backoff sleeping
+    monkeypatch.setattr(ru, "RETRY_BACKOFF_FACTOR", 0)
+    ru.reset_sessions()
+    yield
+    ru.reset_sessions()
+
+
+@pytest.fixture
+def config():
+    return SimpleNamespace(maap_token="abc123", content_type="application/json")
+
+
+@pytest.fixture
+def server():
+    """Local server replying with each queued status in turn (last one repeats)."""
+    state = {"statuses": [200], "calls": [], "headers": []}
+
+    class Handler(BaseHTTPRequestHandler):
+        def _respond(self):
+            state["calls"].append(self.command)
+            state["headers"].append(dict(self.headers))
+            idx = min(len(state["calls"]) - 1, len(state["statuses"]) - 1)
+            length = int(self.headers.get("Content-Length") or 0)
+            self.rfile.read(length)
+            self.send_response(state["statuses"][idx])
+            self.send_header("Content-Length", "2")
+            self.end_headers()
+            self.wfile.write(b"ok")
+
+        do_GET = do_POST = do_PUT = do_DELETE = _respond
+
+        def log_message(self, *args):
+            pass
+
+    httpd = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    state["url"] = f"http://127.0.0.1:{httpd.server_port}/x"
+    yield state
+    httpd.shutdown()
+    httpd.server_close()
+
+
+def test_session_is_shared_and_pooled():
+    assert ru.get_session() is ru.get_session()
+    assert ru.get_session(retry_post=True) is not ru.get_session()
+
+
+def test_retry_policy_methods():
+    get_retry = ru.get_session().get_adapter("https://x").max_retries
+    assert get_retry.total == ru.RETRY_TOTAL
+    assert set(get_retry.status_forcelist) == set(ru.RETRY_STATUS_CODES)
+    assert get_retry.respect_retry_after_header
+    assert get_retry.allowed_methods == ru.IDEMPOTENT_METHODS
+    post_retry = ru.get_session(retry_post=True).get_adapter("https://x").max_retries
+    assert "POST" in post_retry.allowed_methods
+
+
+def test_get_retries_transient_errors(server, config):
+    server["statuses"] = [503, 502, 200]
+    r = ru.request(ru.GET, server["url"], config)
+    assert r.status_code == 200
+    assert server["calls"] == ["GET"] * 3
+
+
+@pytest.mark.parametrize("method", [ru.PUT, ru.DELETE])
+def test_idempotent_methods_retry(server, config, method):
+    server["statuses"] = [429, 200]
+    assert ru.request(method, server["url"], config).status_code == 200
+    assert len(server["calls"]) == 2
+
+
+def test_post_not_retried_by_default(server, config):
+    server["statuses"] = [503, 200]
+    r = ru.request(ru.POST, server["url"], config, json={"a": 1})
+    assert r.status_code == 503
+    assert server["calls"] == ["POST"]
+
+
+def test_post_retried_when_opted_in(server, config):
+    server["statuses"] = [503, 200]
+    r = ru.request(ru.POST, server["url"], config, retry_post=True, json={"a": 1})
+    assert r.status_code == 200
+    assert server["calls"] == ["POST", "POST"]
+
+
+def test_exhausted_retries_return_last_response(server, config):
+    server["statuses"] = [503]
+    r = ru.request(ru.GET, server["url"], config)
+    assert r.status_code == 503
+    assert len(server["calls"]) == ru.RETRY_TOTAL + 1
+
+
+def test_client_errors_not_retried(server, config):
+    server["statuses"] = [404]
+    assert ru.request(ru.GET, server["url"], config).status_code == 404
+    assert len(server["calls"]) == 1
+
+
+def test_default_timeout_applied(monkeypatch, config):
+    seen = {}
+
+    def fake_request(self, **kwargs):
+        seen.update(kwargs)
+        return requests.Response()
+
+    monkeypatch.setattr(requests.Session, "request", fake_request)
+    ru.request("get", "http://example.com", config)
+    assert seen["timeout"] == ru.DEFAULT_TIMEOUT
+    ru.request("get", "http://example.com", config, timeout=5)
+    assert seen["timeout"] == 5
+
+
+def test_token_header_read_per_call(server, config):
+    ru.request(ru.GET, server["url"], config)
+    config.maap_token = "refreshed"
+    ru.request(ru.GET, server["url"], config)
+    assert server["headers"][0]["token"] == "abc123"
+    assert server["headers"][1]["token"] == "refreshed"
+
+
+def test_bearer_token_uses_authorization_header(server, config, monkeypatch):
+    monkeypatch.setenv("MAAP_PGT", "pgt-ticket")
+    config.maap_token = "Bearer xyz"
+    ru.request(ru.GET, server["url"], config)
+    assert server["headers"][0]["Authorization"] == "Bearer xyz"
+    assert "token" not in server["headers"][0]
+    assert server["headers"][0]["proxy-ticket"] == "pgt-ticket"
+
+
+def test_auth_false_sends_no_maap_credentials(server, config, monkeypatch):
+    monkeypatch.setenv("MAAP_PGT", "pgt-ticket")
+    ru.request(ru.GET, server["url"], auth=False)
+    sent = {k.lower() for k in server["headers"][0]}
+    assert not sent & {"token", "authorization", "proxy-ticket"}
+
+
+def test_auth_requires_config():
+    with pytest.raises(ValueError):
+        ru.request(ru.GET, "http://example.com")
+
+
+def test_extra_headers_override_and_are_not_mutated(server, config):
+    extra = {"Accept": "text/plain"}
+    ru.request(ru.GET, server["url"], config, headers=extra)
+    assert server["headers"][0]["Accept"] == "text/plain"
+    assert extra == {"Accept": "text/plain"}
+
+
+def test_tokens_not_logged(server, config, caplog):
+    caplog.set_level("DEBUG", logger=ru.logger.name)
+    ru.request(ru.GET, server["url"], config)
+    assert "abc123" not in caplog.text
+
+
+def test_make_request_supports_all_methods(server, config):
+    for method in (ru.GET, ru.POST, ru.PUT, ru.DELETE):
+        assert ru.make_request(server["url"], config, request_type=method).status_code == 200
+    assert server["calls"] == ["GET", "POST", "PUT", "DELETE"]
+    assert ru.make_dps_request(server["url"], config) == "ok"