"""Magento REST API client - implements MagentoApiClient port.""" from __future__ import annotations import json import logging from urllib.parse import quote, urlencode from typing import Any, Dict, List, Optional, Tuple from magento.oauth_client import MagentoOAuthClient logger = logging.getLogger(__name__) def _rfc3986_quote(s: str, safe: str = "", encoding=None, errors=None) -> str: """RFC 3986 encoding for OAuth (space -> %20, not +). Accepts urlencode quote_via signature.""" return quote(str(s), safe=safe or "", encoding=encoding or "utf-8", errors=errors or "strict") def _sku_path(sku: str) -> str: """Encode SKU for Magento REST path segments. Some Magento stacks behind Apache/cPanel decode ``%2F`` before the request reaches Magento's OAuth validator. When a SKU contains ``/``, signing the once-encoded path can then mismatch the server-side base string and Magento responds with ``The signature is invalid`` on writes. Double-encoding literal slashes keeps the request URI stable through that proxy layer while still resolving the intended SKU inside Magento. """ text = str(sku).replace("/", "%2F") return quote(text, safe="", encoding="utf-8", errors="replace") def _sku_path_candidates(sku: str) -> List[str]: """Path encodings to try for Magento ``/products/{sku}/...`` routes. Prefer double-encoded slashes (OAuth/proxy safe), then single-encoded. Some Magento installs false-404 slash SKUs on the double-encoded form for GET media / product reads even though the entity exists. """ sku_text = str(sku or "").strip() if not sku_text: return [] candidates = [_sku_path(sku_text)] if "/" in sku_text: single = quote(sku_text, safe="", encoding="utf-8", errors="replace") if single not in candidates: candidates.append(single) return candidates def _magento_flag(value: Any) -> int: """Magento REST expects 0/1 ints for most EAV boolean-like attribute flags.""" if isinstance(value, bool): return 1 if value else 0 if value is None: return 0 try: return 1 if int(value) else 0 except (TypeError, ValueError): return 0 class MagentoRestClient: """Magento REST API adapter. Implements MagentoApiClient protocol.""" def __init__(self, oauth_client: MagentoOAuthClient) -> None: self._client = oauth_client def get_product(self, sku: str) -> Optional[Dict[str, Any]]: """GET /V1/products/{sku}. Returns None if 404. Raises on 5xx (do not hide Magento outages). Slash SKUs try alternate path encodings, then searchCriteria exact match — Magento path lookup often false-404s ``/`` SKUs even when the product exists. """ sku_text = str(sku or "").strip() last_status = 404 last_err: Optional[str] = None for path_sku in _sku_path_candidates(sku_text): status, body, err = self._client._request("GET", f"products/{path_sku}") if status == 200 and isinstance(body, dict): return body last_status, last_err = status, err if status >= 500: raise MagentoServerError(sku_text, status, err or f"HTTP {status}") if status not in (400, 404): return None if "/" in sku_text or " " in sku_text: status, items, err = self.search_products(sku=sku_text, page_size=5) if status >= 500: raise MagentoServerError(sku_text, status, err or f"HTTP {status}") if status == 200 and items: wanted = sku_text.upper() for item in items: if isinstance(item, dict) and str(item.get("sku") or "").strip().upper() == wanted: return item if last_status >= 500: raise MagentoServerError(sku_text, last_status, last_err or f"HTTP {last_status}") return None def create_product(self, payload: Dict[str, Any]) -> Dict[str, Any]: status, body, err = self._client._request( "POST", "products", json={"product": payload} ) if status not in (200, 201): raise RuntimeError(err or f"HTTP {status}") return body or {} def update_product(self, sku: str, payload: Dict[str, Any]) -> Dict[str, Any]: status, body, err = self._client._request( "PUT", f"products/{_sku_path(sku)}", json={"product": payload} ) if status == 401 and "/" in str(sku): # Some Magento stacks reject OAuth signatures when the SKU lives in # the path and contains "/". Fall back to ProductRepository::save. status, body, err = self._client._request( "POST", "products", json={"product": payload} ) if status == 404: raise ProductNotFoundError(sku) if status not in (200, 201): raise RuntimeError(err or f"HTTP {status}") return body or {} def put_product( self, sku: str, payload: Dict[str, Any], *, save_options: bool = False, ) -> tuple[int, Optional[Dict[str, Any]]]: """Low-level PUT. Returns (status, body). Use for 404 detection.""" request_body: Dict[str, Any] = {"product": payload} if save_options: request_body["saveOptions"] = True status, body, err = self._client._request( "PUT", f"products/{_sku_path(sku)}", json=request_body ) if status == 401 and "/" in str(sku): status, body, err = self._client._request( "POST", "products", json=request_body ) return status, body def delete_product(self, sku: str) -> Tuple[int, Optional[str]]: """DELETE /V1/products/{sku}. Hard-deletes the product from Magento. Returns (status, error). status 200/204 = success (including already-absent). Slash SKUs try double-encoded then single-encoded path segments. A path 404 is only treated as success when searchCriteria confirms the SKU is gone (avoids false "deleted" when Magento still has the entity but path lookup failed). """ sku_text = str(sku or "").strip() if not sku_text: return 400, "sku required" path_candidates = _sku_path_candidates(sku_text) last_status = 404 last_err: Optional[str] = None for path_sku in path_candidates: status, _, err = self._client._request("DELETE", f"products/{path_sku}") if status in (200, 204): return status, None last_status, last_err = status, err # Keep trying alternate encodings only for not-found / bad-request path issues. if status not in (400, 404): return status, err or f"HTTP {status}" if last_status == 404 and not self.product_exists_by_sku_search(sku_text): return 204, None return last_status, last_err or f"HTTP {last_status}" def product_exists_by_sku_search(self, sku: str) -> bool: """True when Magento searchCriteria finds an exact SKU (works for slash SKUs).""" sku_text = str(sku or "").strip() if not sku_text: return False # Prefer searchCriteria — GET /products/{sku} path encoding can false-404 slash SKUs. status, items, _err = self.search_products(sku=sku_text, page_size=5) if status != 200 or not items: return False wanted = sku_text.upper() for item in items: if isinstance(item, dict) and str(item.get("sku") or "").strip().upper() == wanted: return True return False def post_product( self, payload: Dict[str, Any], *, save_options: bool = False, ) -> tuple[int, Optional[Dict[str, Any]]]: """Low-level POST. Returns (status, body).""" request_body: Dict[str, Any] = {"product": payload} if save_options: request_body["saveOptions"] = True status, body, _ = self._client._request( "POST", "products", json=request_body ) return status, body def get_bundle_options( self, sku: str, ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/bundle-products/{sku}/options/all.""" status, body, err = self._client._request( "GET", f"bundle-products/{_sku_path(sku)}/options/all", ) if status == 404: return status, [], None if status != 200: return status, [], err or f"HTTP {status}" return status, body if isinstance(body, list) else [], None def add_bundle_option( self, option: Dict[str, Any], ) -> Tuple[int, Optional[Any], Optional[str]]: """POST /V1/bundle-products/options/add.""" status, body, err = self._client._request( "POST", "bundle-products/options/add", json={"option": option}, ) if status not in (200, 201): return status, body, err or f"HTTP {status}" return status, body, None def update_bundle_option( self, option_id: int, option: Dict[str, Any], ) -> Tuple[int, Optional[Any], Optional[str]]: """PUT /V1/bundle-products/options/{optionId}.""" status, body, err = self._client._request( "PUT", f"bundle-products/options/{int(option_id)}", json={"option": option}, ) if status not in (200, 201): return status, body, err or f"HTTP {status}" return status, body, None def add_bundle_product_link( self, sku: str, option_id: int, linked_product: Dict[str, Any], ) -> Tuple[int, Optional[Any], Optional[str]]: """POST /V1/bundle-products/{sku}/links/{optionId}.""" status, body, err = self._client._request( "POST", f"bundle-products/{_sku_path(sku)}/links/{int(option_id)}", json={"linkedProduct": linked_product}, ) if status not in (200, 201): return status, body, err or f"HTTP {status}" return status, body, None def delete_bundle_option( self, sku: str, option_id: int, ) -> Tuple[int, Optional[str]]: """DELETE /V1/bundle-products/{sku}/options/{optionId}.""" status, _body, err = self._client._request( "DELETE", f"bundle-products/{_sku_path(sku)}/options/{int(option_id)}", ) if status in (200, 201, 204, 404): return status, None return status, err or f"HTTP {status}" # --- Async bulk product upserts --- def put_products_bulk_async( self, products: List[Dict[str, Any]] ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """PUT /async/bulk/V1/products/bySku — queue product updates. ``products`` is a list of Magento product dicts (each must include ``sku``). Returns (http_status, body_with_bulk_uuid, error). """ if not products: return 200, {"bulk_uuid": None, "request_items": [], "errors": False}, None body_list = [{"product": p} for p in products] url = self._client.async_bulk_url("products/bySku") return self._client._request_url("PUT", url, json=body_list) def post_products_bulk_async( self, products: List[Dict[str, Any]] ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """POST /async/bulk/V1/products — queue product creates. ``products`` is a list of Magento product dicts. Returns (http_status, body_with_bulk_uuid, error). """ if not products: return 200, {"bulk_uuid": None, "request_items": [], "errors": False}, None body_list = [{"product": p} for p in products] url = self._client.async_bulk_url("products") return self._client._request_url("POST", url, json=body_list) def get_bulk_status( self, bulk_uuid: str ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """GET /V1/bulk/{bulkUuid}/status.""" return self._client._request("GET", f"bulk/{bulk_uuid}/status") def get_bulk_detailed_status( self, bulk_uuid: str ) -> tuple[int, Optional[Dict[str, Any]], Optional[str]]: """GET /V1/bulk/{bulkUuid}/detailed-status (includes serialized payloads).""" return self._client._request("GET", f"bulk/{bulk_uuid}/detailed-status") def verify_connection(self) -> bool: """GET store/storeViews - lightweight verify.""" status, body, err = self._client._request("GET", "store/storeViews") return status == 200 and body is not None def list_products( self, *, page_size: int = 20, current_page: int = 1, ) -> tuple[int, Optional[list], Optional[str]]: """ GET /V1/products with searchCriteria to list all products. Returns (status, items_list, total_count, error). """ params = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), } status, body, err = self._client._request("GET", "products", params=params) if status != 200: return status, None, 0, err items = body.get("items", []) if isinstance(body, dict) else [] total = body.get("total_count", len(items)) if isinstance(body, dict) else 0 return status, items, total, None def list_products_updated_since( self, updated_since: str, *, page_size: int = 100, current_page: int = 1, ) -> tuple[int, Optional[list], Optional[int], Optional[str]]: """ GET products filtered by updated_at > value. For daily pull. updated_since: ISO datetime or date string. Returns (status, items, total_count, error). Uses RFC 3986 encoding for OAuth signature compatibility with Magento. """ params = { "searchCriteria[filter_groups][0][filters][0][field]": "updated_at", "searchCriteria[filter_groups][0][filters][0][value]": updated_since, "searchCriteria[filter_groups][0][filters][0][condition_type]": "gt", "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{self._client._url('products')}?{query}" status, body, err = self._client._request_url("GET", url) if status != 200: return status, None, None, err items = body.get("items", []) if isinstance(body, dict) else [] total = body.get("total_count", len(items)) if isinstance(body, dict) else 0 return status, items, total, None def get_products_by_skus(self, skus: list[str]) -> tuple[int, Optional[list], Optional[str]]: """ GET products by exact SKU match. Uses GET /V1/products/{sku} per SKU (no search API) to avoid partial/fuzzy matching where child SKUs like EOB-W18-S could match when querying for parent EOB-W18. Returns (status, items_list, error). """ if not skus: return 200, [], None items: list = [] for sku in skus: s = str(sku).strip() if not s: continue prod = self.get_product(s) if prod is not None: items.append(prod) return 200, items, None def search_products( self, *, sku: Optional[str] = None, entity_id: Optional[int] = None, page_size: int = 20, ) -> tuple[int, Optional[list], Optional[str]]: """ GET /V1/products with exact searchCriteria filters. Useful as a fallback when direct GET /products/{sku} misses a product. """ params: Dict[str, str] = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": "1", } idx = 0 if sku is not None: params[f"searchCriteria[filter_groups][{idx}][filters][0][field]"] = "sku" params[f"searchCriteria[filter_groups][{idx}][filters][0][value]"] = str(sku) params[f"searchCriteria[filter_groups][{idx}][filters][0][condition_type]"] = "eq" idx += 1 if entity_id is not None: params[f"searchCriteria[filter_groups][{idx}][filters][0][field]"] = "entity_id" params[f"searchCriteria[filter_groups][{idx}][filters][0][value]"] = str(int(entity_id)) params[f"searchCriteria[filter_groups][{idx}][filters][0][condition_type]"] = "eq" query = urlencode(params, quote_via=_rfc3986_quote) url = f"{self._client._url('products')}?{query}" status, body, err = self._client._request_url("GET", url) if status != 200: return status, None, err or f"HTTP {status}" items = body.get("items", []) if isinstance(body, dict) else [] return status, items if isinstance(items, list) else [], None def verify_connection_with_details(self) -> dict: """ Verify and return details for debugging. Returns {"ok": bool, "http_status": int, "error": str|None} """ status, body, err = self._client._request("GET", "store/storeViews") return { "ok": status == 200 and body is not None, "http_status": status, "error": err, } # --- Media (base64 upload) --- def get_product_media(self, sku: str) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/products/{sku}/media. Returns (status, entries, error). Slash SKUs retry single-encoded path when double-encoded returns 400/404. """ last_status = 404 last_err: Optional[str] = None for path_sku in _sku_path_candidates(sku): status, body, err = self._client._request("GET", f"products/{path_sku}/media") if status == 200: entries = ( body if isinstance(body, list) else (body.get("entries", []) if isinstance(body, dict) else []) ) return status, entries, None last_status, last_err = status, err if status not in (400, 404): return status, [], err or f"HTTP {status}" return last_status, [], last_err or f"HTTP {last_status}" def post_product_media( self, sku: str, entry: Dict[str, Any] ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """ POST /V1/products/{sku}/media with base64 content. entry must have: media_type, label, position, types, content {base64_encoded_data, type, name} Returns (status, body, error). body contains id on success. """ last_status = 400 last_err: Optional[str] = None for path_sku in _sku_path_candidates(sku): status, body, err = self._client._request( "POST", f"products/{path_sku}/media", json={"entry": entry} ) if status in (200, 201): return status, body, None last_status, last_err = status, err if status not in (400, 404): return status, None, err or f"HTTP {status}" return last_status, None, last_err or f"HTTP {last_status}" def delete_product_media(self, sku: str, entry_id: int) -> Tuple[int, Optional[str]]: """DELETE /V1/products/{sku}/media/{entryId}. Returns (status, error).""" last_status = 404 last_err: Optional[str] = None for path_sku in _sku_path_candidates(sku): status, _, err = self._client._request( "DELETE", f"products/{path_sku}/media/{entry_id}" ) if status in (200, 204): return status, None last_status, last_err = status, err if status not in (400, 404): return status, err or f"HTTP {status}" return last_status, last_err or f"HTTP {last_status}" # --- Configurable relations --- def get_configurable_options( self, sku: str ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/configurable-products/{sku}/options/all. Returns (status, options, error).""" status, body, err = self._client._request( "GET", f"configurable-products/{_sku_path(sku)}/options/all" ) if status != 200: return status, [], err or f"HTTP {status}" opts = body if isinstance(body, list) else [] return status, opts, None def get_configurable_children(self, sku: str) -> Tuple[int, List[str], Optional[str]]: """ GET /V1/configurable-products/{sku}/children. Returns (status, child_skus, error). """ status, body, err = self._client._request( "GET", f"configurable-products/{sku}/children" ) if status != 200: return status, [], err or f"HTTP {status}" items = body if isinstance(body, list) else [] skus = [ str(it.get("sku", "")).strip() for it in items if isinstance(it, dict) and it.get("sku") ] return status, skus, None def add_configurable_child( self, parent_sku: str, child_sku: str ) -> Tuple[int, Optional[str]]: """ Link child to configurable. Tries routes in order; 2xx = success. Routes: /child (body childSku), /children (body childSku), /children/{childSku} (body {}). """ attempts: List[Dict[str, Any]] = [] def try_route(path_suffix: str, json_body: Dict[str, Any]) -> Tuple[int, Optional[str]]: status, body, err = self._client._request( "POST", f"configurable-products/{_sku_path(parent_sku)}/{path_suffix}", json=json_body, ) attempts.append({"path": path_suffix, "status": status, "body": body, "err": err}) if status in (200, 201): return status, None return status, err or f"HTTP {status}" status, err = try_route("child", {"childSku": child_sku}) if err is None: return status, None if status not in (404,) and "route" not in str(err or "").lower() and "not found" not in str(err or "").lower(): return status, err status, err = try_route("children", {"childSku": child_sku}) if err is None: return status, None if status not in (404,) and "route" not in str(err or "").lower() and "not found" not in str(err or "").lower(): return status, err status, err = try_route(f"children/{_sku_path(child_sku)}", {}) if err is None: return status, None raise ConfigurableChildLinkError( parent_sku, child_sku, attempted_routes=[a["path"] for a in attempts], attempts_detail=attempts, ) def remove_configurable_child( self, parent_sku: str, child_sku: str ) -> Tuple[int, Optional[str]]: """ Remove child from configurable. DELETE /V1/configurable-products/{parentSku}/children/{childSku} Returns (status, error). """ status, _, err = self._client._request( "DELETE", f"configurable-products/{_sku_path(parent_sku)}/children/{_sku_path(child_sku)}", ) if status in (200, 204): return status, None return status, err or f"HTTP {status}" def create_configurable_option( self, parent_sku: str, payload: Dict[str, Any] ) -> Tuple[int, Optional[Dict[str, Any]]]: """POST /V1/configurable-products/{sku}/options. Returns (status, body).""" status, body, err = self._client._request( "POST", f"configurable-products/{_sku_path(parent_sku)}/options", json={"option": payload}, ) if status not in (200, 201): return status, body if isinstance(body, dict) else {"error": err} return status, body def delete_configurable_option( self, parent_sku: str, option_id: int ) -> Tuple[int, Optional[str]]: """ DELETE /V1/configurable-products/{sku}/options/{optionId}. Removes a configurable option (axis) from the product. Returns (status, error). Use when replacing wrong axes with correct ones. """ status, _, err = self._client._request( "DELETE", f"configurable-products/{_sku_path(parent_sku)}/options/{option_id}", ) if status in (200, 204): return status, None return status, err or f"HTTP {status}" # --- Attribute sets --- def list_attribute_sets( self, *, page_size: int = 100, current_page: int = 1, entity_type_id: Optional[int] = 4, ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """ GET /V1/products/attribute-sets/sets/list with searchCriteria. entity_type_id=4 filters for catalog_product. If filter returns empty, retries without filter (some Magento versions ignore or reject the filter). Caller should filter by entity_type_id. Returns (status, items_list, error). items may include entity_type_id for client-side filtering. """ params: Dict[str, str] = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), "searchCriteria[sortOrders][0][field]": "attribute_set_id", "searchCriteria[sortOrders][0][direction]": "ASC", } if entity_type_id is not None: params["searchCriteria[filter_groups][0][filters][0][field]"] = "entity_type_id" params["searchCriteria[filter_groups][0][filters][0][value]"] = str(entity_type_id) params["searchCriteria[filter_groups][0][filters][0][condition_type]"] = "eq" status, body, err = self._client._request( "GET", "products/attribute-sets/sets/list", params=params ) if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None items = body.get("items", []) if isinstance(body, dict) else [] items = items if isinstance(items, list) else [] if not items and entity_type_id is not None: params2 = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), "searchCriteria[sortOrders][0][field]": "attribute_set_id", "searchCriteria[sortOrders][0][direction]": "ASC", } status2, body2, err2 = self._client._request( "GET", "products/attribute-sets/sets/list", params=params2 ) if status2 == 200: items = body2.get("items", []) if isinstance(body2, dict) else (body2 if isinstance(body2, list) else []) return status, items if isinstance(items, list) else [], None def get_attributes_for_set( self, attribute_set_id: int ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """ GET /V1/products/attribute-sets/{id}/attributes. Returns (status, items_list, error). items have attribute_id, attribute_code, etc. """ status, body, err = self._client._request( "GET", f"products/attribute-sets/{attribute_set_id}/attributes" ) if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None if isinstance(body, dict) and "items" in body: items = body.get("items", []) return status, items if isinstance(items, list) else [], None return status, [], None def get_attribute_id_by_code(self, attribute_code: str) -> Optional[int]: """GET /V1/products/attributes/{code}. Returns attribute_id or None if not found.""" status, detail = self.get_attribute_detail(attribute_code) return detail.get("attribute_id") if status == 200 and detail else None def get_attribute_detail( self, attribute_code: str ) -> Tuple[int, Optional[Dict[str, Any]]]: """ GET /V1/products/attributes/{code}. Returns (status, body). body has attribute_id, frontend_label, frontend_input, etc. """ status, body, _ = self._client._request( "GET", f"products/attributes/{attribute_code}" ) if status != 200 or not isinstance(body, dict): return status, None aid = body.get("attribute_id") if aid is None: return status, None label = body.get("frontend_label") or body.get("default_frontend_label") return status, { "attribute_id": int(aid), "attribute_code": str(attribute_code), "frontend_label": str(label).strip() if label else attribute_code.replace("_", " ").title(), "frontend_input": body.get("frontend_input"), "backend_type": body.get("backend_type"), "is_user_defined": body.get("is_user_defined"), } def get_attribute_options(self, attribute_code: str) -> Tuple[int, List[Dict[str, Any]]]: """ GET /V1/products/attributes/{code}/options. Returns (status, options). options: [{"label": "Black", "value": "49"}, ...]. value is value_index. Fallback: try full attribute GET and parse options from it. """ status, body, err = self._client._request( "GET", f"products/attributes/{attribute_code}/options" ) if status == 200 and isinstance(body, list): return status, body if status == 404 or "not found" in str(err or "").lower(): attr_status, attr_body, _ = self._client._request( "GET", f"products/attributes/{attribute_code}" ) if attr_status == 200 and isinstance(attr_body, dict): opts = attr_body.get("options") or attr_body.get("option") or [] if isinstance(opts, list): return 200, opts if isinstance(opts, dict): return 200, [opts] return status, [] def create_attribute( self, attribute_code: str, *, frontend_label: str, frontend_input: str = "select", backend_type: str = "int", is_required: bool = False, is_user_defined: bool = True, is_global: int = 1, is_searchable: bool = True, is_visible_in_advanced_search: bool = True, is_filterable: bool = True, is_filterable_in_search: bool = True, is_used_for_promo_rules: bool = False, is_html_allowed_on_front: bool = False, used_in_product_listing: bool = True, used_for_sort_by: bool = False, is_configurable: bool = True, scope: str = "global", ) -> Tuple[int, Optional[Dict[str, Any]]]: """ POST /V1/products/attributes. Create a new product attribute. Returns (status, body). body has attribute_id on success. Magento 2.4.8 validates payload strictly against ProductAttributeInterface — only send fields with a defined setter. is_global, is_configurable, and scope are all rejected on create; scope defaults to global, select attributes are usable as configurables by default. """ del scope, is_global, is_configurable # not in ProductAttributeInterface; rejected on create attribute: Dict[str, Any] = { "attribute_code": attribute_code, "frontend_input": frontend_input, "backend_type": backend_type, "default_frontend_label": frontend_label, "is_required": _magento_flag(is_required), "is_user_defined": _magento_flag(is_user_defined), } if frontend_input == "select": attribute["is_searchable"] = _magento_flag(is_searchable) attribute["is_visible_in_advanced_search"] = _magento_flag(is_visible_in_advanced_search) attribute["is_filterable"] = _magento_flag(is_filterable) attribute["is_filterable_in_search"] = _magento_flag(is_filterable_in_search) attribute["is_used_for_promo_rules"] = _magento_flag(is_used_for_promo_rules) attribute["used_in_product_listing"] = _magento_flag(used_in_product_listing) attribute["used_for_sort_by"] = _magento_flag(used_for_sort_by) elif frontend_input == "boolean": attribute["used_in_product_listing"] = _magento_flag(used_in_product_listing) else: attribute["is_html_allowed_on_front"] = _magento_flag(is_html_allowed_on_front) attribute["used_in_product_listing"] = _magento_flag(used_in_product_listing) payload: Dict[str, Any] = {"attribute": attribute} status, body, err = self._client._request( "POST", "products/attributes", json=payload ) if status not in (200, 201): if isinstance(body, dict): return status, body return status, {"error": err or str(body)} return status, body if isinstance(body, dict) else None def get_attribute(self, attribute_code: str) -> Tuple[int, Optional[Dict[str, Any]]]: """GET /V1/products/attributes/{attributeCode}.""" status, body, _ = self._client._request( "GET", f"products/attributes/{attribute_code}" ) return status, body if status == 200 and isinstance(body, dict) else None def update_attribute( self, attribute_code: str, payload: Dict[str, Any], ) -> Tuple[int, Optional[Dict[str, Any]]]: """PUT /V1/products/attributes/{attributeCode} — partial attribute update.""" status, body, _ = self._client._request( "PUT", f"products/attributes/{attribute_code}", json={"attribute": payload}, ) return status, body if status in (200, 201) and isinstance(body, dict) else body def delete_attribute( self, attribute_code: str, ) -> Tuple[int, Optional[Any], Optional[str]]: """DELETE /V1/products/attributes/{attributeCode}.""" status, body, err = self._client._request( "DELETE", f"products/attributes/{attribute_code}", ) if status not in (200, 201, 204): return status, body, err or f"HTTP {status}" return status, body, None def create_attribute_option( self, attribute_code: str, label: str, *, sort_order: int = 0 ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """ POST /V1/products/attributes/{attributeCode}/options. Add option to select attribute. Returns (status, body, error). body has value (value_index) on success. """ payload = {"option": {"label": label, "sort_order": sort_order}} status, body, err = self._client._request( "POST", f"products/attributes/{attribute_code}/options", json=payload, ) if status not in (200, 201): return status, None, err or str(body) if isinstance(body, dict): return status, body, None if isinstance(body, (int, float)) or (isinstance(body, str) and str(body).isdigit()): return status, {"value": int(body)}, None return status, None, None def list_attribute_set_groups( self, attribute_set_id: int ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/products/attribute-sets/groups/list filtered by attribute_set_id.""" params: Dict[str, str] = { "searchCriteria[filter_groups][0][filters][0][field]": "attribute_set_id", "searchCriteria[filter_groups][0][filters][0][value]": str(attribute_set_id), "searchCriteria[filter_groups][0][filters][0][condition_type]": "eq", "searchCriteria[pageSize]": "50", "searchCriteria[currentPage]": "1", } status, body, err = self._client._request( "GET", "products/attribute-sets/groups/list", params=params ) if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None items = body.get("items", []) if isinstance(body, dict) else [] return status, items if isinstance(items, list) else [], None @staticmethod def _pick_attribute_group_id(groups: List[Dict[str, Any]]) -> Optional[int]: preferred = ("general", "product details", "attributes", "product information") for name in preferred: for group in groups: if not isinstance(group, dict): continue group_name = str(group.get("attribute_group_name") or "").strip().lower() if group_name == name: gid = group.get("attribute_group_id") or group.get("id") if gid is not None: return int(gid) for group in groups: if not isinstance(group, dict): continue gid = group.get("attribute_group_id") or group.get("id") if gid is not None: return int(gid) return None def add_attribute_to_attribute_set( self, attribute_set_id: int, attribute_code: str, *, attribute_group_id: Optional[int] = None, ) -> Tuple[int, Optional[str]]: """ Assign attribute to attribute set. Magento 2 REST typically uses: POST /V1/products/attribute-sets/attributes Body: { attributeSetId, attributeGroupId, attributeCode, sortOrder } If attribute_group_id not provided, uses first group of the set. Returns (status, error). """ if attribute_group_id is None: g_status, grp_list, g_err = self.list_attribute_set_groups(attribute_set_id) if g_status != 200 or not grp_list: g_status, groups_body, _ = self._client._request( "GET", f"products/attribute-sets/{attribute_set_id}/groups", ) if g_status == 200 and groups_body: if isinstance(groups_body, list): grp_list = groups_body elif isinstance(groups_body, dict): grp_list = groups_body.get("items", []) or [] if not grp_list: return g_status or 500, g_err or "Could not get attribute set groups" attribute_group_id = self._pick_attribute_group_id(grp_list) if attribute_group_id is None: return 400, "Attribute set has no groups" payload = { "attributeSetId": attribute_set_id, "attributeGroupId": attribute_group_id, "attributeCode": attribute_code, "sortOrder": 0, } status, body, err = self._client._request( "POST", "products/attribute-sets/attributes", json=payload, ) if status in (200, 201): return status, None return status, err or str(body) def find_attribute_by_frontend_label( self, label: str ) -> Tuple[int, Optional[Dict[str, Any]]]: """ GET /V1/products/attributes with searchCriteria by frontend_label. Returns (status, best_match_dict). Best match = case-insensitive equals. """ params = { "searchCriteria[filter_groups][0][filters][0][field]": "frontend_label", "searchCriteria[filter_groups][0][filters][0][value]": f"%{label}%", "searchCriteria[filter_groups][0][filters][0][condition_type]": "like", "searchCriteria[pageSize]": "50", "searchCriteria[currentPage]": "1", } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{self._client._url('products/attributes')}?{query}" status, body, _ = self._client._request_url("GET", url) if status != 200 or not isinstance(body, dict): return status, None items = body.get("items", []) if isinstance(body.get("items"), list) else [] label_lower = label.strip().lower() for it in items: if not isinstance(it, dict): continue fl = (it.get("frontend_label") or it.get("default_frontend_label") or "").strip().lower() if fl == label_lower or fl == label.strip(): return status, it if items: return status, items[0] return status, None # --- Store info --- def get_store_websites(self) -> Tuple[int, List[Dict[str, Any]]]: """GET /V1/store/websites. Returns (status, websites_list).""" status, body, _ = self._client._request("GET", "store/websites") if status != 200 or not isinstance(body, list): return status, [] return status, body def get_store_views(self) -> Tuple[int, List[Dict[str, Any]]]: """GET /V1/store/storeViews. Returns (status, store_views_list).""" status, body, _ = self._client._request("GET", "store/storeViews") if status != 200 or not isinstance(body, list): return status, [] return status, body def get_store_groups(self) -> Tuple[int, List[Dict[str, Any]]]: """GET /V1/store/storeGroups. Returns (status, groups_list).""" status, body, _ = self._client._request("GET", "store/storeGroups") if status != 200 or not isinstance(body, list): return status, [] return status, body # --- Inventory (MSI) --- def get_inventory_sources(self) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/inventory/sources. Returns (status, sources_list, error). MSI module required.""" status, body, err = self._client._request("GET", "inventory/sources") if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None if isinstance(body, dict) and "items" in body: return status, body.get("items", []), None return status, [], None def get_inventory_stocks(self) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/inventory/stocks. Returns (status, stocks_list, error). MSI module required.""" status, body, err = self._client._request("GET", "inventory/stocks") if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None if isinstance(body, dict) and "items" in body: return status, body.get("items", []), None return status, [], None def get_sources_assigned_to_stock( self, stock_id: int ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """ GET /V1/inventory/get-sources-assigned-to-stock-ordered-by-priority/{stockId}. Returns (status, sources_list, error). """ status, body, err = self._client._request( "GET", f"inventory/get-sources-assigned-to-stock-ordered-by-priority/{stock_id}" ) if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, body, None if isinstance(body, dict) and "items" in body: return status, body.get("items", []), None # Some Magento versions return a flat list of source objects if isinstance(body, dict) and "source_code" in body: return status, [body], None return status, [], None def post_source_items( self, source_items: List[Dict[str, Any]] ) -> Tuple[int, Optional[str]]: """ POST /V1/inventory/source-items. Assign products to MSI source. source_items: [{ "sku": "...", "source_code": "...", "quantity": 0, "status": 1 }] status: 0=out of stock, 1=in stock. Returns (status, error). """ if not source_items: return 200, None status, body, err = self._client._request( "POST", "inventory/source-items", json={"sourceItems": source_items} ) if status in (200, 201): return status, None return status, err or (str(body) if body else f"HTTP {status}") def get_source_items_for_skus( self, skus: List[str] ) -> Tuple[int, Dict[str, List[Dict[str, Any]]], Optional[str]]: """ GET /V1/inventory/source-items filtered by SKU(s). Returns (status, {sku: [source_items]}, error). Each source_item has: sku, source_code, quantity, status. """ if not skus: return 200, {}, None skus = [str(s).strip() for s in skus if str(s).strip()] if not skus: return 200, {}, None values = ",".join(skus) if len(skus) > 1 else skus[0] cond = "in" if len(skus) > 1 else "eq" params = { "searchCriteria[filter_groups][0][filters][0][field]": "sku", "searchCriteria[filter_groups][0][filters][0][value]": values, "searchCriteria[filter_groups][0][filters][0][condition_type]": cond, "searchCriteria[pageSize]": str(max(len(skus) * 10, 100)), "searchCriteria[currentPage]": "1", } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{self._client._url('inventory/source-items')}?{query}" status, body, err = self._client._request_url("GET", url) if status != 200: return status, {}, err or f"HTTP {status}" items = [] if isinstance(body, list): items = body elif isinstance(body, dict) and "items" in body: items = body.get("items", []) by_sku: Dict[str, List[Dict[str, Any]]] = {} for item in items: s = item.get("sku", "") if s: by_sku.setdefault(s, []).append(item) for s in skus: by_sku.setdefault(s, []) return status, by_sku, None # --- Categories --- def get_categories_tree(self) -> Tuple[int, Optional[Dict[str, Any]]]: """GET /V1/categories. Returns (status, tree_dict).""" status, body, _ = self._client._request("GET", "categories") if status != 200 or not isinstance(body, dict): return status, None return status, body def create_category(self, payload: Dict[str, Any]) -> Tuple[int, Optional[Dict[str, Any]]]: """POST /V1/categories. Returns (status, body).""" status, body, err = self._client._request( "POST", "categories", json={"category": payload} ) if status not in (200, 201): return status, {"error": err or f"HTTP {status}"} return status, body def search_categories( self, *, category_id: Optional[int] = None, name: Optional[str] = None, parent_id: Optional[int] = None, url_key: Optional[str] = None, ) -> Tuple[int, Optional[List[Dict[str, Any]]]]: """GET categories/list with searchCriteria. Returns (status, items_list).""" params = {"searchCriteria[pageSize]": "50"} idx = 0 prefix = "searchCriteria[filterGroups]" # Magento REST uses camelCase if category_id is not None: params[f"{prefix}[{idx}][filters][0][field]"] = "entity_id" params[f"{prefix}[{idx}][filters][0][value]"] = str(int(category_id)) params[f"{prefix}[{idx}][filters][0][condition_type]"] = "eq" idx += 1 if name is not None: params[f"{prefix}[{idx}][filters][0][field]"] = "name" params[f"{prefix}[{idx}][filters][0][value]"] = str(name) params[f"{prefix}[{idx}][filters][0][condition_type]"] = "eq" idx += 1 if parent_id is not None: params[f"{prefix}[{idx}][filters][0][field]"] = "parent_id" params[f"{prefix}[{idx}][filters][0][value]"] = str(parent_id) params[f"{prefix}[{idx}][filters][0][condition_type]"] = "eq" idx += 1 if url_key is not None: params[f"{prefix}[{idx}][filters][0][field]"] = "url_key" params[f"{prefix}[{idx}][filters][0][value]"] = str(url_key) params[f"{prefix}[{idx}][filters][0][condition_type]"] = "eq" idx += 1 query = urlencode(params, quote_via=_rfc3986_quote) full_url = f"{self._client._url('categories/list')}?{query}" status, body, _ = self._client._request_url("GET", full_url) if status != 200 or not isinstance(body, dict): return status, None items = body.get("items", []) return status, items if isinstance(items, list) else None def get_category(self, category_id: int) -> Tuple[int, Optional[Dict[str, Any]]]: """GET /V1/categories/{id}.""" status, body, _ = self._client._request("GET", f"categories/{int(category_id)}") if status != 200 or not isinstance(body, dict): return status, None return status, body def update_category(self, category_id: int, payload: Dict[str, Any]) -> Tuple[int, Optional[Dict[str, Any]]]: """PUT /V1/categories/{id}.""" status, body, err = self._client._request( "PUT", f"categories/{int(category_id)}", json={"category": payload}, ) if status not in (200, 201): return status, {"error": err or f"HTTP {status}", "body": body} return status, body if isinstance(body, dict) else {"id": category_id} def delete_category(self, category_id: int) -> Tuple[int, Optional[Any], Optional[str]]: """DELETE /V1/categories/{id}.""" status, body, err = self._client._request("DELETE", f"categories/{int(category_id)}") if status not in (200, 201, 204): return status, body, err or f"HTTP {status}" return status, body, None def get_category_products(self, category_id: int) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET products currently assigned to one Magento category.""" status, body, err = self._client._request("GET", f"categories/{int(category_id)}/products") if status != 200: return status, [], err or f"HTTP {status}" return status, body if isinstance(body, list) else [], None def assign_product_to_category( self, category_id: int, sku: str, *, position: int = 0, ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """POST one product/category link through Magento's categoryLink API.""" status, body, err = self._client._request( "POST", f"categories/{int(category_id)}/products", json={ "productLink": { "sku": str(sku), "category_id": str(int(category_id)), "position": int(position), } }, ) if status not in (200, 201): return status, body if isinstance(body, dict) else None, err or f"HTTP {status}" return status, body if isinstance(body, dict) else {"success": bool(body)}, None def remove_product_from_category( self, category_id: int, sku: str, ) -> Tuple[int, Optional[str]]: """DELETE one product/category link through Magento's categoryLink API.""" status, _, err = self._client._request( "DELETE", f"categories/{int(category_id)}/products/{_sku_path(sku)}", ) if status in (200, 204): return status, None return status, err or f"HTTP {status}" def get_product_links( self, sku: str, link_type: str ) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /V1/products/{sku}/links/{linkType} (related|upsell|crosssell).""" status, body, err = self._client._request( "GET", f"products/{_sku_path(sku)}/links/{quote(str(link_type), safe='')}", ) if status == 404: return status, [], None if status != 200: return status, [], err or f"HTTP {status}" if isinstance(body, list): return status, [item for item in body if isinstance(item, dict)], None return status, [], None def assign_product_links( self, sku: str, items: List[Dict[str, Any]] ) -> Tuple[int, Optional[Any], Optional[str]]: """POST /V1/products/{sku}/links — assign related/upsell/crosssell links.""" if not items: return 200, True, None status, body, err = self._client._request( "POST", f"products/{_sku_path(sku)}/links", json={"items": items}, ) if status not in (200, 201): return status, body, err or f"HTTP {status}" return status, body, None def remove_product_link( self, sku: str, link_type: str, linked_product_sku: str ) -> Tuple[int, Optional[str]]: """DELETE /V1/products/{sku}/links/{type}/{linkedSku}.""" status, _, err = self._client._request( "DELETE", f"products/{_sku_path(sku)}/links/" f"{quote(str(link_type), safe='')}/{_sku_path(linked_product_sku)}", ) if status in (200, 204): return status, None return status, err or f"HTTP {status}" def upload_collection_asset( self, *, filename: str, mime_type: str, base64_content: str, file_size_bytes: Optional[int] = None, ) -> Tuple[int, Optional[Dict[str, Any]]]: """Upload an image/PDF into Magento ``pub/media/piehs/collections``.""" payload = { "filename": filename, "mimeType": mime_type, "base64Content": base64_content, } raw_bytes = ( int(file_size_bytes) if file_size_bytes is not None else (len(base64_content) * 3) // 4 ) post_json_bytes = len(json.dumps(payload, ensure_ascii=False).encode("utf-8")) logger.info( "Magento collection asset upload: filename=%s mime=%s file_bytes=%s " "post_json_bytes=%s (%.2f MB) url=%s", filename, mime_type, raw_bytes, post_json_bytes, post_json_bytes / (1024 * 1024), self._client._url("piehs/collection-media"), ) status, body, err = self._client._request( "POST", "piehs/collection-media", json=payload, ) if status not in (200, 201): logger.error( "Magento collection asset upload failed: status=%s filename=%s " "file_bytes=%s post_json_bytes=%s (%.2f MB) url=%s", status, filename, raw_bytes, post_json_bytes, post_json_bytes / (1024 * 1024), self._client._url("piehs/collection-media"), ) return status, {"error": err or f"HTTP {status}", "body": body} if isinstance(body, str): return status, {"url": body} return status, body if isinstance(body, dict) else None def get_category_attribute(self, attribute_code: str) -> Tuple[int, Optional[Dict[str, Any]]]: """GET /V1/categories/attributes/{code}.""" status, body, _ = self._client._request("GET", f"categories/attributes/{attribute_code}") if status != 200 or not isinstance(body, dict): return status, None return status, body def create_category_attribute( self, attribute_code: str, *, frontend_label: str, frontend_input: str = "textarea", backend_type: str = "text", is_required: bool = False, is_user_defined: bool = True, ) -> Tuple[int, Optional[Dict[str, Any]]]: """POST /V1/categories/attributes. Note: Magento core does not expose this route (returns 404). Category attributes must be installed via a Magento data patch; see ``magento.collection_landing_setup``. """ attribute: Dict[str, Any] = { "attribute_code": attribute_code, "frontend_input": frontend_input, "backend_type": backend_type, "default_frontend_label": frontend_label, "is_required": _magento_flag(is_required), "is_user_defined": _magento_flag(is_user_defined), } status, body, err = self._client._request( "POST", "categories/attributes", json={"attribute": attribute}, ) if status not in (200, 201): return status, {"error": err or f"HTTP {status}", "body": body} return status, body def update_product_media( self, sku: str, entry_id: int, entry: Dict[str, Any] ) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """PUT /V1/products/{sku}/media/{entryId}. Returns (status, body, error).""" last_status = 400 last_err: Optional[str] = None for path_sku in _sku_path_candidates(sku): status, body, err = self._client._request( "PUT", f"products/{path_sku}/media/{entry_id}", json={"entry": entry} ) if status in (200, 201): return status, body, None last_status, last_err = status, err if status not in (400, 404): return status, None, err or f"HTTP {status}" return last_status, None, last_err or f"HTTP {last_status}" #GET /rest/V1/cmsPage/search #GET /rest/V1/cmsPage/search?searchCriteria[currentPage]=1&searchCriteria[pageSize]=100 #TypeError: magento.magento_api.MagentoRestClient.search_cms_pages() argument after ** must be a mapping, not NoneType def search_cms_pages(self, *, page_size: int = 100, current_page: int = 1, **search_criteria: Optional[Dict[str, Any]]) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /rest/V1/cmsPage/search. Returns (status, pages_list, error).""" params = {"searchCriteria[pageSize]": page_size, "searchCriteria[currentPage]": current_page} if search_criteria: params.update(search_criteria) status, body, err = self._client._request("GET", "cmsPage/search", params=params) if status != 200: return status, [], err or f"HTTP {status}" return status, body if isinstance(body, list) else [], None #GET /rest/V1/cmsPage/1 def get_cms_page(self, page_id: int) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """GET /rest/V1/cmsPage/{id}. Returns (status, page_dict, error).""" status, body, err = self._client._request("GET", f"cmsPage/{page_id}") if status != 200: return status, None, err or f"HTTP {status}" return status, body if isinstance(body, dict) else None, None #GET /rest/V1/cmsPage/1 #searchCriteria[sortOrders][0][field]=identifier&searchCriteria[sortOrders][0][direction]=asc def get_cms_pages(self, *, page_size: int = 100, current_page: int = 1) -> Tuple[int, List[Dict[str, Any]], Optional[str]]: """GET /rest/V1/cmsPage. Returns (status, pages_list, error).""" params = {"searchCriteria[pageSize]": page_size, "searchCriteria[currentPage]": current_page, "searchCriteria[sortOrders][0][direction]": "asc"} status, body, err = self._client._request("GET", "cmsPage/search", params=params) if status != 200: return status, [], err or f"HTTP {status}" return status, body if isinstance(body, list) else [], None def find_cms_page_by_identifier(self, identifier: str) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """Find one CMS page by identifier.""" params = { "searchCriteria[pageSize]": "1", "searchCriteria[currentPage]": "1", "searchCriteria[filterGroups][0][filters][0][field]": "identifier", "searchCriteria[filterGroups][0][filters][0][value]": str(identifier), "searchCriteria[filterGroups][0][filters][0][condition_type]": "eq", } status, body, err = self._client._request("GET", "cmsPage/search", params=params) if status != 200: return status, None, err or f"HTTP {status}" items = body.get("items", []) if isinstance(body, dict) else body if isinstance(body, list) else [] if not items: return status, None, None return status, items[0] if isinstance(items[0], dict) else None, None def create_cms_page(self, payload: Dict[str, Any]) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """POST /V1/cmsPage. Returns (status, body, error).""" status, body, err = self._client._request("POST", "cmsPage", json={"page": payload}) if status not in (200, 201): return status, body if isinstance(body, dict) else None, err or f"HTTP {status}" return status, body if isinstance(body, dict) else {"id": body}, None def update_cms_page(self, page_id: int, payload: Dict[str, Any]) -> Tuple[int, Optional[Dict[str, Any]], Optional[str]]: """PUT /V1/cmsPage/{id}. Returns (status, body, error).""" status, body, err = self._client._request("PUT", f"cmsPage/{int(page_id)}", json={"page": payload}) if status not in (200, 201): return status, body if isinstance(body, dict) else None, err or f"HTTP {status}" return status, body if isinstance(body, dict) else {"id": page_id}, None class ProductNotFoundError(RuntimeError): def __init__(self, sku: str) -> None: self.sku = sku super().__init__(f"Product not found: {sku}") class MagentoServerError(RuntimeError): """Magento 5xx - do not treat as 'not found'.""" def __init__(self, sku: str, status: int, message: str) -> None: self.sku = sku self.status = status self.message = message super().__init__(f"Magento server error for {sku}: HTTP {status} - {message}") class ConfigurableChildLinkError(RuntimeError): """All child-link routes failed.""" def __init__( self, parent_sku: str, child_sku: str, *, attempted_routes: list, attempts_detail: Optional[List[Dict[str, Any]]] = None, last_status: Optional[int] = None, last_error: Optional[str] = None, ) -> None: self.parent_sku = parent_sku self.child_sku = child_sku self.attempted_routes = attempted_routes self.attempts_detail = attempts_detail or [] last = self.attempts_detail[-1] if self.attempts_detail else {} self.last_status = last_status or last.get("status", 0) self.last_error = last_error or last.get("err") or str(last.get("body", "")) super().__init__( f"Could not link {child_sku} to {parent_sku}: " f"tried {attempted_routes}, last HTTP {self.last_status}: {self.last_error}" )