From d59cb76a688fbc9c456aa6c7931a8411b9a7c3a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 17:57:51 +0000 Subject: [PATCH] feat: add shared retrying requests session client Add a requests.Session-based client to maap.utils.requests_utils with retries (429/5xx + connection errors, idempotent methods only unless opted in), default timeouts, request-time auth injection with an auth=False opt-out, and token redaction in debug logs. make_request and make_dps_request now delegate to it and support PUT/DELETE. No other callers are migrated yet. Refs #137 Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Rp9gSo9Fw6M7DUFhWnNPaU --- CHANGELOG.md | 2 + maap/utils/requests_utils.py | 137 ++++++++++++++++++++++++---- test/test_requests_utils.py | 172 +++++++++++++++++++++++++++++++++++ 3 files changed, 295 insertions(+), 16 deletions(-) create mode 100644 test/test_requests_utils.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bef2003..760155d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/maap/utils/requests_utils.py b/maap/utils/requests_utils.py index 7a53a74..4021f98 100644 --- a/maap/utils/requests_utils.py +++ b/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 @@ class HTTPMethod(Enum): 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,6 +100,10 @@ def generate_dps_headers(config: MaapConfig, content_type=None): 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, @@ -44,24 +111,62 @@ def check_response(dps_response): 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, diff --git a/test/test_requests_utils.py b/test/test_requests_utils.py new file mode 100644 index 0000000..9844fd3 --- /dev/null +++ b/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"