"""Payoneer API connector.""" import contextlib from datetime import datetime, timezone from email.utils import parsedate_to_datetime import json import os import time from typing import Any import httpx from pydantic import ValidationError from tenacity import ( retry_if_exception_type, RetryCallState, RetryError, Retrying, stop_after_attempt, ) from config import ( app_logger as logger, ENVIRONMENT, PROD_ENVIRONMENT, QA_ENVIRONMENT, UAT_ENVIRONMENT, ) from src.connectors import secrets from src.models import PayoneerPayeeDetails PAYONEER_API_URL = os.environ.get( 'PAYONEER_API_URL', 'https://api.sandbox.payoneer.com' ) # Auth env vars are read inside `_get_auth_token` / the 401 refresh branch, # not at import time, so changes (test monkeypatch, runtime overrides) take # effect without re-importing the module. PAYONEER_GET_PAYEE_DETAILS_ENDPOINT = ( '/v4/programs/{program_id}/payees/{payee_id}/details' ) MAX_RETRIES = 3 BACKOFF_BASE_SECONDS = 1.0 BACKOFF_MAX_SECONDS = 30.0 # 4xx status codes that are permanent - retrying with the same request will # never succeed, so we fail fast instead of wasting time. PERMANENT_4XX_CODES = {400, 403, 422} # Exception types that tenacity will retry on. Anything else (e.g. our own # PayoneerApiException for permanent errors) propagates immediately. TRANSIENT_EXCEPTIONS = (httpx.HTTPError, json.JSONDecodeError) # Indirection so tests can patch the sleep used by tenacity. Tenacity captures # its sleep default at import time, so patching tenacity.nap.sleep later has # no effect on Retrying instances. _sleep = time.sleep REMOTE_SECRET_ENVIRONMENTS = {QA_ENVIRONMENT, UAT_ENVIRONMENT, PROD_ENVIRONMENT} # Payoneer issues 28-char opaque tokens (exclusive of the `Bearer ` prefix # we add ourselves in `_build_headers`). A shorter local value is a # placeholder / accidental copy — fail fast rather than send garbage to # the API and chase a cryptic 401 later. MIN_DEV_TOKEN_LENGTH = 28 class PayoneerApiException(Exception): """Payoneer API exception.""" def _get_auth_token() -> str: """Retrieve the Payoneer auth token. In QA/UAT/Prod, fetches from Secrets Manager (cached per container). In dev, uses the PAYONEER_AUTH_TOKEN environment variable. """ if ENVIRONMENT in REMOTE_SECRET_ENVIRONMENTS: arn = os.environ.get('PAYONEER_AUTH_TOKEN_ARN') if not arn: raise PayoneerApiException('PAYONEER_AUTH_TOKEN_ARN is not configured') return secrets.get_secret(arn) # Strip once and compare: whitespace from copy-paste / `.env` files # would otherwise slip past the length check and be forwarded into # the `Bearer ` header. token = (os.environ.get('PAYONEER_AUTH_TOKEN') or '').strip() if len(token) < MIN_DEV_TOKEN_LENGTH: raise PayoneerApiException('No valid Payoneer auth token configured') return token def _build_headers() -> dict[str, str]: """Build Payoneer API request headers.""" return {'Authorization': f'Bearer {_get_auth_token()}'} def _parse_retry_after(value: str | None) -> float | None: """Parse the Retry-After header. RFC 9110 allows delta-seconds or HTTP-date; support both, returning seconds-from-now. """ if not value: return None with contextlib.suppress(ValueError): return float(value) with contextlib.suppress(TypeError, ValueError): dt = parsedate_to_datetime(value) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return (dt - datetime.now(timezone.utc)).total_seconds() return None def _wait_payoneer(retry_state: RetryCallState) -> float: """Compute backoff delay for a failed attempt. Honors Retry-After on HTTPStatusError responses (clamped to [0, BACKOFF_MAX_SECONDS]), otherwise exponential backoff: 1s, 2s, 4s, ... """ exc = retry_state.outcome.exception() if retry_state.outcome else None if isinstance(exc, httpx.HTTPStatusError): retry_after = _parse_retry_after(exc.response.headers.get('Retry-After')) if retry_after is not None: # Clamp non-negative: servers can send 0 or garbage; sleep() would # raise on negatives. return max(0.0, min(retry_after, BACKOFF_MAX_SECONDS)) delay = BACKOFF_BASE_SECONDS * float(2 ** (retry_state.attempt_number - 1)) return min(delay, BACKOFF_MAX_SECONDS) def get_payee_details( program_id: int, payee_id: int | str ) -> PayoneerPayeeDetails | None: """Fetch payee details with payout methods from Payoneer API. GET /v4/programs/{program_id}/payees/{payee_id}/details Returns the validated PayoneerPayeeDetails model, or None if the payee is not found (404). Raises PayoneerApiException on permanent errors or after retries exhausted. Validation happens at this boundary so downstream code never sees raw dicts. Retries on: - Network errors (connection, timeout) - 5xx server errors - 429 Too Many Requests (honors Retry-After header) - Malformed JSON responses (via json.JSONDecodeError) Fails fast on: - 404 (returns None) - Permanent 4xx errors (400, 403, 422) - retrying won't help - Payoneer error payload in a 200 response body 401 is retried once after invalidating the cached auth token, to handle rotation on warm containers. """ endpoint = PAYONEER_GET_PAYEE_DETAILS_ENDPOINT.format( program_id=program_id, payee_id=payee_id ) url = f'{PAYONEER_API_URL}{endpoint}' params = {'show_payout_method_details': 'true'} # Capture refresh-state via closure so we only refresh the cached auth # token once per request — a second 401 after refresh is treated as a # permanent failure rather than burning remaining retries on bad creds. auth_refreshed = False def log_attempt(attempt_number: int, exc: BaseException | None) -> None: logger.warning( f'Payoneer API attempt {attempt_number}/{MAX_RETRIES} ' f'failed for program={program_id} payee={payee_id}: {exc}' ) def fetch() -> dict[str, Any] | None: # Headers built each attempt so 401-triggered cache invalidation # below takes effect on the next retry (secrets.invalidate drops # the cached token; the next _get_auth_token() call re-fetches # from Secrets Manager). nonlocal auth_refreshed headers = _build_headers() response = httpx.get(url, headers=headers, params=params, timeout=30) if response.status_code == 404: return None # 401 — token may have been rotated on a warm container. Refresh # once, then treat subsequent 401s as permanent. if response.status_code == 401: if auth_refreshed: raise PayoneerApiException( f'Payoneer auth failed (401) after token refresh for ' f'program={program_id} payee={payee_id}' ) auth_refreshed = True arn = os.environ.get('PAYONEER_AUTH_TOKEN_ARN') if arn: try: secrets.invalidate(arn) except Exception as exc: # botocore.ClientError (or any Secrets Manager failure) # is non-transient: surface as PayoneerApiException so # the processor's per-row handler catches it instead # of an un-typed exception escaping through # future.result() and aborting the whole batch. raise PayoneerApiException( f'Failed to refresh Payoneer auth token for ' f'program={program_id} payee={payee_id}: {exc}' ) from exc raise httpx.HTTPStatusError( f'Payoneer auth failed (401) for program={program_id} payee={payee_id}', request=response.request, response=response, ) # Permanent 4xx errors - fail fast, retrying won't help. if response.status_code in PERMANENT_4XX_CODES: raise PayoneerApiException( f'Payoneer client error {response.status_code} for ' f'program={program_id} payee={payee_id}' ) # 429 (rate limit) - wrap in HTTPStatusError so tenacity retries, # and surface Retry-After via the response for _wait_payoneer. if response.status_code == 429: raise httpx.HTTPStatusError( f'Payoneer rate limited (429) for ' f'program={program_id} payee={payee_id}', request=response.request, response=response, ) # 5xx handled via raise_for_status (retried as HTTPStatusError). response.raise_for_status() result: dict[str, Any] = response.json() if result.get('error'): raise PayoneerApiException( f'Payoneer error for program={program_id} ' f'payee={payee_id}: ' f'{result.get("error_description", result["error"])}' ) # A 200 with neither `error` nor `result` is an upstream contract # violation — do NOT downgrade to "not found" (None) or the caller # treats real data loss as a benign miss. if 'result' not in result: raise PayoneerApiException( f'Payoneer 200 response missing both "error" and "result" keys ' f'for program={program_id} payee={payee_id}' ) payee: dict[str, Any] = result['result'] return payee def before_sleep(retry_state: RetryCallState) -> None: exc = retry_state.outcome.exception() if retry_state.outcome else None log_attempt(retry_state.attempt_number, exc) retryer = Retrying( sleep=_sleep, stop=stop_after_attempt(MAX_RETRIES), wait=_wait_payoneer, retry=retry_if_exception_type(TRANSIENT_EXCEPTIONS), before_sleep=before_sleep, ) try: result = retryer(fetch) except RetryError as e: last = e.last_attempt.exception() # before_sleep only fires between attempts, so the final failed # attempt is not logged by the hook — log it here to preserve the # per-attempt warning trail from the old loop. log_attempt(e.last_attempt.attempt_number, last) raise PayoneerApiException( f'Failed after {MAX_RETRIES} retries for program={program_id} ' f'payee={payee_id}: {last}' ) from last if result is None: return None try: return PayoneerPayeeDetails.model_validate(result) except ValidationError as e: # Keep validation failures at row-level blast radius: the processor # catches PayoneerApiException and logs+skips, whereas an uncaught # ValidationError would escape future.result() and abort the batch. raise PayoneerApiException( f'Payoneer response failed validation for ' f'program={program_id} payee={payee_id}: {e}' ) from e