"""OAuth1 HTTP client for Magento REST API.""" from __future__ import annotations import logging import time import warnings from typing import Any, Dict, Mapping, Optional import requests from urllib3.exceptions import InsecureRequestWarning from requests_oauthlib import OAuth1 # Immediate retry config: on connection error or 5xx, retry this many times # with a short delay within a single recovery round. _IMMEDIATE_RETRY_COUNT = 3 _IMMEDIATE_RETRY_DELAY = 30.0 # seconds between immediate retries # After a full immediate-retry round still sees server-down, wait these # intervals (seconds) then run another round. Covers Magento deploy restarts # (often 5–10 minutes) before giving up on the request. _SERVER_DOWN_RECOVERY_WAITS = ( 5 * 60, # 5 minutes 10 * 60, # 10 minutes 30 * 60, # 30 minutes ) logger = logging.getLogger(__name__) class MagentoOAuthClient: """OAuth1-signed HTTP client for Magento REST.""" def __init__( self, *, base_url: str, consumer_key: str, consumer_secret: str, access_token: str, access_token_secret: str, rest_base_path: str = "V1", signature_method: str = "HMAC-SHA256", timeout: float = 60.0, default_headers: Optional[Mapping[str, str]] = None, fallback_base_url: Optional[str] = None, fallback_headers: Optional[Mapping[str, str]] = None, verify_ssl: bool = True, fallback_verify_ssl: bool = True, max_recovery_rounds: Optional[int] = None, ) -> None: self.base_url = base_url.rstrip("/") self.rest_base_path = rest_base_path.strip("/") self.timeout = timeout self.default_headers = dict(default_headers or {}) self.fallback_base_url = (fallback_base_url or "").rstrip("/") self.fallback_headers = dict(fallback_headers or {}) self.verify_ssl = bool(verify_ssl) self.fallback_verify_ssl = bool(fallback_verify_ssl) # None = full server-down recovery (3 immediate + long waits). Jobs that # hammer Magento should pass 1 to fail fast instead of sleeping for hours. self.max_recovery_rounds = max_recovery_rounds # Magento 2.4.4+ requires HMAC-SHA256; HMAC-SHA1 is no longer supported. use_sha256 = "HMAC-SHA256" self._auth = OAuth1( client_key=consumer_key, client_secret=consumer_secret, resource_owner_key=access_token, resource_owner_secret=access_token_secret, signature_method=use_sha256, ) def _url(self, path: str) -> str: if path.startswith("/"): path = path[1:] return f"{self.base_url}/{self.rest_base_path}/{path}" def _fallback_url(self, path: str) -> Optional[str]: if not self.fallback_base_url: return None if path.startswith("/"): path = path[1:] return f"{self.fallback_base_url}/{self.rest_base_path}/{path}" def async_bulk_url(self, resource_path: str) -> str: """Build PaaS async bulk URL: insert ``async/bulk`` before ``V1``. Examples (base_url already includes ``/rest``): - rest_base_path ``V1`` + ``products`` → ``.../async/bulk/V1/products`` - rest_base_path ``us_eng/V1`` + ``products/bySku`` → ``.../us_eng/async/bulk/V1/products/bySku`` """ rbp = self.rest_base_path.strip("/") resource = resource_path.lstrip("/") if rbp == "V1": bulk_base = "async/bulk/V1" elif rbp.endswith("/V1"): store = rbp[: -len("/V1")].strip("/") bulk_base = f"{store}/async/bulk/V1" if store else "async/bulk/V1" else: parts = rbp.split("/") if parts and parts[-1].upper().startswith("V"): head = "/".join(parts[:-1]) bulk_base = f"{head}/async/bulk/{parts[-1]}" if head else f"async/bulk/{parts[-1]}" else: bulk_base = f"async/bulk/{rbp}" return f"{self.base_url}/{bulk_base}/{resource}" def _perform_request( self, method: str, url: str, *, json: Optional[Any] = None, params: Optional[Dict[str, str]] = None, headers: Optional[Mapping[str, str]] = None, auth: Optional[Any] = None, verify_ssl: Optional[bool] = None, ) -> tuple[int, Optional[Dict[str, Any]], Optional[str], requests.Response]: merged_headers = dict(self.default_headers) merged_headers.update(headers or {}) request_kwargs: Dict[str, Any] = { "auth": auth or self._auth, "timeout": self.timeout, "headers": merged_headers or None, "verify": self.verify_ssl if verify_ssl is None else bool(verify_ssl), } if json is not None: request_kwargs["json"] = json if params is not None: request_kwargs["params"] = params verify = bool(request_kwargs.get("verify", True)) if verify: resp = requests.request(method, url, **request_kwargs) else: with warnings.catch_warnings(): warnings.simplefilter("ignore", InsecureRequestWarning) resp = requests.request(method, url, **request_kwargs) status = resp.status_code try: body = resp.json() if resp.content else None except Exception: body = None err = None if status < 400 else (resp.text or str(status)) return status, body, err, resp def _request_url( self, method: str, full_url: str, **kwargs: Any ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """Make request to pre-built URL. Use for OAuth-compatible query encoding.""" try: status, body, err, _ = self._perform_request( method, full_url, json=kwargs.pop("json", None), params=kwargs.pop("params", None), headers=kwargs.pop("headers", None), ) return status, body, err except requests.RequestException as e: logger.exception("Magento API request failed: %s %s", method, full_url[:80]) return 0, None, str(e) except Exception as e: logger.exception("Magento API request failed: %s %s", method, full_url[:80]) return 0, None, str(e) def _request( self, method: str, path: str, *, json: Optional[Any] = None, params: Optional[Dict[str, str]] = None, ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """Returns (status_code, response_body, error_message). On connection errors (status 0) or 5xx responses, runs rounds of ``_IMMEDIATE_RETRY_COUNT`` attempts (``_IMMEDIATE_RETRY_DELAY`` apart). Between rounds, sleeps per ``_SERVER_DOWN_RECOVERY_WAITS`` (5m → 10m → 30m) so Magento deploy restarts can finish before the request is abandoned. """ url = self._url(path) last_result: tuple[int, Optional[Dict[str, Any]], Optional[str]] = ( 0, None, "No attempts made", ) total_rounds = 1 + len(_SERVER_DOWN_RECOVERY_WAITS) if self.max_recovery_rounds is not None: total_rounds = max(1, min(total_rounds, int(self.max_recovery_rounds))) for round_idx in range(total_rounds): if round_idx > 0: wait_secs = _SERVER_DOWN_RECOVERY_WAITS[round_idx - 1] wait_label = f"{wait_secs // 60}m" if wait_secs % 60 == 0 else f"{wait_secs}s" logger.warning( "Magento API still down after round %d/%d — sleeping %s before recovery round %d: %s %s", round_idx, total_rounds, wait_label, round_idx + 1, method, path, ) time.sleep(wait_secs) for attempt in range(_IMMEDIATE_RETRY_COUNT): if attempt > 0: logger.warning( "Magento API retry %d/%d (round %d/%d) in %ds: %s %s", attempt + 1, _IMMEDIATE_RETRY_COUNT, round_idx + 1, total_rounds, int(_IMMEDIATE_RETRY_DELAY), method, path, ) time.sleep(_IMMEDIATE_RETRY_DELAY) try: status, body, err, resp = self._perform_request( method, url, json=json, params=params, verify_ssl=self.verify_ssl, ) last_result = (status, body, err) if status == 413: logger.error( "Magento API HTTP 413: %s %s final_url=%s response_headers=%s body_prefix=%s", method, url, resp.url, {k: v for k, v in resp.headers.items()}, (resp.text or "")[:200], ) fallback_url = self._fallback_url(path) if fallback_url and fallback_url != url: logger.warning( "Retrying Magento API request via internal fallback after 413: %s %s fallback_url=%s", method, path, fallback_url, ) fallback_status, fallback_body, fallback_err, fallback_resp = self._perform_request( method, fallback_url, json=json, params=params, headers=self.fallback_headers, verify_ssl=self.fallback_verify_ssl, ) last_result = (fallback_status, fallback_body, fallback_err) if fallback_status != 413: return fallback_status, fallback_body, fallback_err logger.error( "Magento API fallback also returned 413: %s %s final_url=%s response_headers=%s body_prefix=%s", method, fallback_url, fallback_resp.url, {k: v for k, v in fallback_resp.headers.items()}, (fallback_resp.text or "")[:200], ) status = fallback_status # Only retry on server-down conditions; 4xx are definitive answers. if status == 0 or status >= 500: logger.warning( "Magento API server-down response (attempt %d/%d, round %d/%d): %s %s -> %d", attempt + 1, _IMMEDIATE_RETRY_COUNT, round_idx + 1, total_rounds, method, path, status, ) continue return status, body, err except requests.RequestException as e: logger.warning( "Magento API request failed (attempt %d/%d, round %d/%d): %s %s - %s", attempt + 1, _IMMEDIATE_RETRY_COUNT, round_idx + 1, total_rounds, method, path, e, ) last_result = (0, None, str(e)) except Exception as e: logger.warning( "Magento API request failed (attempt %d/%d, round %d/%d): %s %s - %s", attempt + 1, _IMMEDIATE_RETRY_COUNT, round_idx + 1, total_rounds, method, path, e, ) last_result = (0, None, str(e)) logger.warning( "Magento API recovery round %d/%d exhausted: %s %s", round_idx + 1, total_rounds, method, path, ) logger.error( "Magento API all %d recovery rounds exhausted (%d attempts each): %s %s", total_rounds, _IMMEDIATE_RETRY_COUNT, method, path, ) return last_result def _is_insecure_local_magento_url(url: str) -> bool: """True for loopback / private HTTP(S) Magento transports that use self-signed or broken TLS.""" raw = str(url or "").strip() if not raw: return False try: from urllib.parse import urlparse parsed = urlparse(raw if "://" in raw else f"https://{raw}") except Exception: return False host = (parsed.hostname or "").strip().lower() return host in {"127.0.0.1", "localhost", "::1"} or host.startswith("10.") or host.startswith("192.168.") def build_magento_oauth_kwargs( connection: Any, *, rest_base_path_override: Optional[str] = None, skip_store_code: bool = False, timeout: float = 60.0, verify_ssl: Optional[bool] = None, ) -> Dict[str, Any]: def read(name: str, default: Any = None) -> Any: if isinstance(connection, dict): return connection.get(name, default) return getattr(connection, name, default) store_base_url = str(read("store_base_url") or "").rstrip("/") public_api_base = str(read("magento_api_base_url") or "").rstrip("/") or f"{store_base_url}/rest" internal_api_base = str(read("internal_api_base_url") or "").rstrip("/") or "" prefer_internal_api = bool(read("prefer_internal_api", False) and internal_api_base) base_url = internal_api_base if prefer_internal_api else public_api_base rest_base_path = str(rest_base_path_override or read("rest_base_path") or "V1").strip("/") store_code = str(read("store_code") or "").strip("/") # Bundle option writes must use admin scope (store_id=0). Callers can pass # rest_base_path_override="V1" with skip_store_code=True. if ( not skip_store_code and store_code and not rest_base_path.startswith(f"{store_code}/") ): rest_base_path = f"{store_code}/{rest_base_path}" default_headers: Dict[str, str] = {} if prefer_internal_api: host_header = str(read("internal_host_header") or "").strip() if host_header: default_headers["Host"] = host_header fallback_base_url = None fallback_headers: Dict[str, str] = {} if not prefer_internal_api and internal_api_base: fallback_base_url = internal_api_base host_header = str(read("internal_host_header") or "").strip() if host_header: fallback_headers["Host"] = host_header # Prefer-internal and loopback HTTPS both need verify off (self-signed / broken local TLS). effective_verify = not prefer_internal_api and not _is_insecure_local_magento_url(base_url) if verify_ssl is not None: effective_verify = bool(verify_ssl) fallback_verify = False if fallback_base_url else True if fallback_base_url and _is_insecure_local_magento_url(fallback_base_url): fallback_verify = False return { "base_url": base_url, "consumer_key": read("consumer_key") or "", "consumer_secret": read("consumer_secret") or "", "access_token": read("access_token") or "", "access_token_secret": read("access_token_secret") or "", "rest_base_path": rest_base_path, "signature_method": read("signature_method") or "HMAC-SHA256", "timeout": timeout, "default_headers": default_headers, "fallback_base_url": fallback_base_url, "fallback_headers": fallback_headers, "verify_ssl": effective_verify, "fallback_verify_ssl": fallback_verify, }