""" Push Magento metadata / product / category / inventory snapshots into the Cloudflare Worker cache from the Magento-side environment. Examples: python -m app.jobs.magento_worker_cache_push --connection-id 4 --metadata python -m app.jobs.magento_worker_cache_push --connection-id 4 --sku ACH-B12 --sku ACH-B15 python -m app.jobs.magento_worker_cache_push --connection-id 4 --product-page 1 --product-page-size 25 python -m app.jobs.magento_worker_cache_push --connection-id 4 --category-tree --category-page 1 python -m app.jobs.magento_worker_cache_push --connection-id 4 --metadata --category-tree --product-page 1 --inventory-sku ACH-B12 """ from __future__ import annotations import argparse import json import logging import time from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlencode import requests from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from settings import load_magento_worker_warm_push_config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _rfc3986_quote(s: str, safe: str = "", encoding=None, errors=None) -> str: from urllib.parse import quote return quote(str(s), safe=safe or "", encoding=encoding or "utf-8", errors=errors or "strict") def _normalize_skus(values: List[str]) -> List[str]: out: List[str] = [] seen = set() for value in values: sku = str(value or "").strip() if not sku: continue key = sku.upper() if key in seen: continue seen.add(key) out.append(sku) return out def _post_worker_json(worker_base_url: str, internal_token: str, path: str, payload: Dict[str, Any], timeout_s: int) -> Dict[str, Any]: response = requests.post( f"{worker_base_url}{path}", headers={ "Authorization": f"Bearer {internal_token}", "Content-Type": "application/json", }, json=payload, timeout=timeout_s, ) text = response.text or "" if response.status_code >= 400: raise RuntimeError(f"POST {path} failed: HTTP {response.status_code} {text[:800]}") return response.json() if text.strip() else {"ok": True} def _fetch_products_page(api: MagentoRestClient, *, current_page: int, page_size: int) -> Tuple[int, Dict[str, Any], Optional[str]]: params: Dict[str, str] = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), "searchCriteria[sortOrders][0][field]": "entity_id", "searchCriteria[sortOrders][0][direction]": "ASC", } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{api._client._url('products')}?{query}" status, body, err = api._client._request_url("GET", url) return status, body if isinstance(body, dict) else {}, err def _fetch_categories_page(api: MagentoRestClient, *, current_page: int, page_size: int) -> Tuple[int, Dict[str, Any], Optional[str]]: params: Dict[str, str] = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(current_page), "searchCriteria[sortOrders][0][field]": "entity_id", "searchCriteria[sortOrders][0][direction]": "ASC", } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{api._client._url('categories/list')}?{query}" status, body, err = api._client._request_url("GET", url) return status, body if isinstance(body, dict) else {}, err def _connection_and_api(connection_id: int): from db.magento_repositories import SqlAlchemyMagentoConnectionRepository from db.session import get_session with get_session() as session: conn = SqlAlchemyMagentoConnectionRepository(session).get_for_sync(connection_id) if not conn: raise RuntimeError(f"Magento connection {connection_id} not found") oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) return conn, MagentoRestClient(oauth) def _build_product_publish_items(items: Any) -> List[Dict[str, Any]]: publish_items: List[Dict[str, Any]] = [] for item in items if isinstance(items, list) else []: if not isinstance(item, dict): continue sku = str(item.get("sku") or "").strip() if not sku: continue publish_items.append({"sku": sku, "body": item}) return publish_items def _run_product_page_pushes( api: MagentoRestClient, *, worker_base_url: str, internal_token: str, timeout_s: int, start_page: int, end_page: int, page_size: int, delay_s: float, ) -> List[Dict[str, Any]]: results: List[Dict[str, Any]] = [] for page in range(start_page, end_page + 1): status, body, err = _fetch_products_page(api, current_page=page, page_size=page_size) if status != 200: raise RuntimeError(f"GET /products page {page} failed: HTTP {status} {err}") publish_items = _build_product_publish_items(body.get("items", []) if isinstance(body, dict) else []) worker_result = _post_worker_json( worker_base_url, internal_token, "/internal/cache/products", {"items": publish_items}, timeout_s, ) results.append( { "page": page, "page_size": page_size, "count": len(publish_items), "worker": worker_result, } ) logger.info("[worker-cache-push] pushed product page=%s count=%s", page, len(publish_items)) if delay_s > 0 and page < end_page: time.sleep(delay_s) return results def _run_category_page_pushes( api: MagentoRestClient, *, worker_base_url: str, internal_token: str, timeout_s: int, start_page: int, end_page: int, page_size: int, delay_s: float, ) -> List[Dict[str, Any]]: results: List[Dict[str, Any]] = [] for page in range(start_page, end_page + 1): status, body, err = _fetch_categories_page(api, current_page=page, page_size=page_size) if status != 200: raise RuntimeError(f"GET /categories/list page {page} failed: HTTP {status} {err}") params = { "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(page), "searchCriteria[sortOrders][0][field]": "entity_id", "searchCriteria[sortOrders][0][direction]": "ASC", } worker_result = _post_worker_json( worker_base_url, internal_token, "/internal/cache/categories", { "items": [ { "cache_key": f"categories_page:{page}", "request_path": f"/rest/V1/categories/list?{urlencode(params, quote_via=_rfc3986_quote)}", "body": body, } ] }, timeout_s, ) count = len(body.get("items", []) if isinstance(body, dict) else []) results.append( { "page": page, "page_size": page_size, "count": count, "worker": worker_result, } ) logger.info("[worker-cache-push] pushed category page=%s count=%s", page, count) if delay_s > 0 and page < end_page: time.sleep(delay_s) return results def run_worker_cache_push( connection_id: int, *, metadata: bool, product_skus: List[str], product_page: Optional[int], product_pages: Optional[int], product_page_size: Optional[int], category_tree: bool, category_page: Optional[int], category_pages: Optional[int], category_page_size: Optional[int], inventory_skus: List[str], delay_s: float, ) -> Dict[str, Any]: cfg = load_magento_worker_warm_push_config() if not cfg.internal_token: raise RuntimeError("MAGENTO_WORKER_INTERNAL_TOKEN is required") _, api = _connection_and_api(connection_id) summary: Dict[str, Any] = {"connection_id": connection_id, "worker_base_url": cfg.worker_base_url} if metadata: modules_status, modules_body, modules_err = api._client._request("GET", "modules") if modules_status != 200: raise RuntimeError(f"GET /modules failed: HTTP {modules_status} {modules_err}") websites_status, websites = api.get_store_websites() if websites_status != 200: raise RuntimeError(f"GET /store/websites failed: HTTP {websites_status}") store_views_status, store_views = api.get_store_views() if store_views_status != 200: raise RuntimeError(f"GET /store/storeViews failed: HTTP {store_views_status}") result = _post_worker_json( cfg.worker_base_url, cfg.internal_token, "/internal/cache", { "modules": modules_body, "store_views": store_views, "websites": websites, }, cfg.request_timeout_s, ) summary["metadata"] = result normalized_skus = _normalize_skus(product_skus) if normalized_skus: items = [] for sku in normalized_skus: product = api.get_product(sku) if product is None: logger.warning("Product not found for SKU %s; skipping publish", sku) continue items.append({"sku": sku, "body": product}) if items: result = _post_worker_json( cfg.worker_base_url, cfg.internal_token, "/internal/cache/products", {"items": items}, cfg.request_timeout_s, ) summary["products_by_sku"] = result if product_page is not None: page_size = max(1, int(product_page_size or cfg.default_product_page_size)) start_page = max(1, int(product_page)) total_pages = max(1, int(product_pages or 1)) end_page = start_page + total_pages - 1 summary["product_pages"] = _run_product_page_pushes( api, worker_base_url=cfg.worker_base_url, internal_token=cfg.internal_token, timeout_s=cfg.request_timeout_s, start_page=start_page, end_page=end_page, page_size=page_size, delay_s=max(0.0, float(delay_s)), ) category_publish_items: List[Dict[str, Any]] = [] if category_tree: status, tree = api.get_categories_tree() if status != 200 or not tree: raise RuntimeError(f"GET /categories failed: HTTP {status}") category_publish_items.append({ "cache_key": "categories:tree", "request_path": "/rest/V1/categories", "body": tree, }) if category_page is not None: page_size = max(1, int(category_page_size or cfg.default_category_page_size)) start_page = max(1, int(category_page)) total_pages = max(1, int(category_pages or 1)) end_page = start_page + total_pages - 1 summary["category_pages"] = _run_category_page_pushes( api, worker_base_url=cfg.worker_base_url, internal_token=cfg.internal_token, timeout_s=cfg.request_timeout_s, start_page=start_page, end_page=end_page, page_size=page_size, delay_s=max(0.0, float(delay_s)), ) if category_publish_items: result = _post_worker_json( cfg.worker_base_url, cfg.internal_token, "/internal/cache/categories", {"items": category_publish_items}, cfg.request_timeout_s, ) summary["categories"] = result normalized_inventory_skus = _normalize_skus(inventory_skus) if normalized_inventory_skus: status, by_sku, err = api.get_source_items_for_skus(normalized_inventory_skus) if status != 200: raise RuntimeError(f"GET /inventory/source-items failed: HTTP {status} {err}") publish_items = [] for sku, entries in by_sku.items(): for entry in entries: publish_items.append({ "sku": sku, "source_code": entry.get("source_code"), "quantity": entry.get("quantity"), "status": entry.get("status"), "body": entry, }) result = _post_worker_json( cfg.worker_base_url, cfg.internal_token, "/internal/cache/inventory", {"items": publish_items}, cfg.request_timeout_s, ) summary["inventory"] = result return summary def main() -> None: parser = argparse.ArgumentParser(description="Push Magento warm-cache snapshots into the Worker") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--metadata", action="store_true", help="Push modules/storeViews/websites") parser.add_argument("--sku", action="append", default=[], help="Push a specific product SKU (repeatable)") parser.add_argument("--product-page", type=int, help="Push one Magento product page by searchCriteria page") parser.add_argument("--product-pages", type=int, default=1, help="How many sequential product pages to push starting at --product-page") parser.add_argument("--product-page-size", type=int, help="Page size for --product-page") parser.add_argument("--category-tree", action="store_true", help="Push the full Magento category tree") parser.add_argument("--category-page", type=int, help="Push one Magento categories/list page") parser.add_argument("--category-pages", type=int, default=1, help="How many sequential category pages to push starting at --category-page") parser.add_argument("--category-page-size", type=int, help="Page size for --category-page") parser.add_argument("--inventory-sku", action="append", default=[], help="Push inventory snapshots for SKU(s)") parser.add_argument("--delay-seconds", type=float, default=0.0, help="Delay between sequential page pushes") args = parser.parse_args() result = run_worker_cache_push( args.connection_id, metadata=args.metadata, product_skus=list(args.sku or []), product_page=args.product_page, product_pages=args.product_pages, product_page_size=args.product_page_size, category_tree=args.category_tree, category_page=args.category_page, category_pages=args.category_pages, category_page_size=args.category_page_size, inventory_skus=list(args.inventory_sku or []), delay_s=args.delay_seconds, ) print(json.dumps(result, indent=2, default=str)) if __name__ == "__main__": main()