""" Magento Baseline Store Sync — pulls stores, categories, attribute sets, attributes, and options. CLI: python -m app.jobs.magento_baseline_sync --connection-id X [--force] The product sync worker gates on a fresh baseline run. Attribute-set aware: pulls attribute sets, determines default set (env or "Default" or first), pulls attributes for that set, then fetches detail+options for each (select/multiselect). Env: MAGENTO_BASELINE_TTL_HOURS=24 MAGENTO_DEFAULT_ATTRIBUTE_SET_NAME=Default MAGENTO_DEFAULT_SOURCE_CODE=... # MSI source for new products (e.g. "default", "homesurplus") """ from __future__ import annotations import argparse import logging import os import re import sys import time from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _norm_option_label(label: str) -> str: s = str(label).strip().lower() return re.sub(r"\s+", " ", s) if s else "" def _flatten_category_tree( node: Dict[str, Any], out: List[dict], parent_path_names: str, *, seen_ids: Optional[set] = None, ) -> None: name = str(node.get("name", "")).strip() cat_id = node.get("id") if not cat_id: return if seen_ids is None: seen_ids = set() try: cat_id_int = int(cat_id) except (TypeError, ValueError): return path = str(node.get("path", "") or "") path_names = f"{parent_path_names}/{name}" if parent_path_names else name if cat_id_int not in seen_ids: seen_ids.add(cat_id_int) out.append({ "category_id": cat_id_int, "parent_id": node.get("parent_id"), "name": name, "path": path, "path_names": path_names, "level": node.get("level", 0), "is_active": node.get("is_active", True), }) for child in node.get("children_data", []): if isinstance(child, dict): _flatten_category_tree(child, out, path_names, seen_ids=seen_ids) SELECT_OR_MULTISELECT = ("select", "multiselect") def _raise_api_error(label: str, status: int, err: Optional[str] = None) -> None: """Raise with the underlying transport/API message when available (HTTP 0 = no response).""" detail = (err or "").strip() or f"HTTP {status}" raise RuntimeError(f"{label} failed: {detail}") def _request_list( api: Any, method: str, path: str ) -> tuple[int, list, Optional[str]]: """Call Magento REST and normalize list responses.""" status, body, err = api._client._request(method, path) if status != 200: return status, [], err if isinstance(body, list): return status, body, None return status, [], err or f"HTTP {status}: expected JSON array" def run_baseline_sync( connection_id: int, *, force: bool = False, session=None, ) -> Dict[str, Any]: """ Execute baseline sync for a connection: stores, categories, attributes, options. Returns result dict with counts and status. """ from db.session import get_session from db.magento_repositories import ( SqlAlchemyMagentoConnectionRepository, SqlAlchemyMagentoStoreRegistryRepository, SqlAlchemyCategoryRegistryRepository, SqlAlchemyAttributeRegistryRepository, SqlAlchemyAttributeSetRegistryRepository, SqlAlchemyAttributeSetAttributesRepository, SqlAlchemySourceRegistryRepository, SqlAlchemyStockRegistryRepository, SqlAlchemyStockSourceLinksRepository, SqlAlchemyBaselineSyncRunRepository, ) from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from magento.magento_api import MagentoRestClient from settings import load_magento_baseline_config, load_magento_sync_runtime_config baseline_cfg = load_magento_baseline_config() runtime_cfg = load_magento_sync_runtime_config() api_delay_s = max(0, runtime_cfg.api_delay_ms) / 1000.0 def _delay() -> None: if api_delay_s > 0: time.sleep(api_delay_s) own_session = session is None ctx = get_session() if own_session else _noop_ctx(session) with ctx as sess: conn_repo = SqlAlchemyMagentoConnectionRepository(sess) store_repo = SqlAlchemyMagentoStoreRegistryRepository(sess) cat_repo = SqlAlchemyCategoryRegistryRepository(sess) attr_repo = SqlAlchemyAttributeRegistryRepository(sess) attr_set_repo = SqlAlchemyAttributeSetRegistryRepository(sess) attr_set_attrs_repo = SqlAlchemyAttributeSetAttributesRepository(sess) source_repo = SqlAlchemySourceRegistryRepository(sess) stock_repo = SqlAlchemyStockRegistryRepository(sess) stock_source_repo = SqlAlchemyStockSourceLinksRepository(sess) baseline_repo = SqlAlchemyBaselineSyncRunRepository(sess) if not force and baseline_repo.is_fresh(connection_id, baseline_cfg.ttl_hours): last = baseline_repo.get_last_successful(connection_id) logger.info("Baseline is fresh (TTL %dh). Last run: %s", baseline_cfg.ttl_hours, last.get("finished_at") if last else "?") return {"status": "skipped", "reason": "fresh", "last_run": last} conn = conn_repo.get_for_sync(connection_id) if not conn: return {"status": "failed", "error": f"Connection {connection_id} not found"} oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) api_root = f"{conn['magento_api_base_url'].rstrip('/')}/{conn['rest_base_path'].strip('/')}" now = datetime.now(timezone.utc) run_id = baseline_repo.create_run(connection_id, now) sess.commit() counts: Dict[str, Any] = {"stores": 0, "categories": 0, "attribute_sets": 0, "attributes": 0, "options": 0} try: logger.info("[baseline] Magento API root: %s", api_root) preflight = api.verify_connection_with_details() if not preflight.get("ok"): _raise_api_error( "Magento API preflight (GET /store/storeViews)", int(preflight.get("http_status") or 0), preflight.get("error"), ) # 1) Stores / websites logger.info("[baseline] Pulling store websites...") _delay() ws_status, websites, ws_err = _request_list(api, "GET", "store/websites") if ws_status != 200: _raise_api_error("GET /store/websites", ws_status, ws_err) _delay() sv_status, store_views, sv_err = _request_list(api, "GET", "store/storeViews") if sv_status != 200: _raise_api_error("GET /store/storeViews", sv_status, sv_err) _delay() sg_status, store_groups, _ = _request_list(api, "GET", "store/storeGroups") groups_by_id = {} if sg_status == 200 and store_groups: groups_by_id = {g["id"]: g for g in store_groups if isinstance(g, dict) and "id" in g} store_rows: List[Dict[str, Any]] = [] views_by_group: Dict[int, List[dict]] = {} for sv in store_views: if isinstance(sv, dict) and "store_group_id" in sv: views_by_group.setdefault(sv["store_group_id"], []).append(sv) for ws in websites: if not isinstance(ws, dict): continue web_id = ws.get("id") if web_id is None: continue web_code = ws.get("code", "") web_name = ws.get("name", "") default_group_id = ws.get("default_group_id") group = groups_by_id.get(default_group_id) if default_group_id else None views = views_by_group.get(default_group_id, []) if default_group_id else [] if views: for sv in views: store_rows.append({ "website_id": int(web_id), "website_code": web_code, "website_name": web_name, "store_group_id": default_group_id, "store_group_name": group.get("name") if group else None, "store_id": sv.get("id"), "store_code": sv.get("code"), "store_name": sv.get("name"), "is_default": bool(ws.get("id") == 1), "raw_json": {"website": ws, "store_view": sv, "group": group}, }) else: store_rows.append({ "website_id": int(web_id), "website_code": web_code, "website_name": web_name, "store_group_id": default_group_id, "store_group_name": group.get("name") if group else None, "store_id": None, "store_code": None, "store_name": None, "is_default": bool(ws.get("id") == 1), "raw_json": {"website": ws, "group": group}, }) counts["stores"] = store_repo.bulk_replace(connection_id, store_rows, now) logger.info("[baseline] Stored %d store registry entries", counts["stores"]) # 2) Categories logger.info("[baseline] Pulling category tree...") _delay() cat_status, tree = api.get_categories_tree() if cat_status != 200 or not tree: raise RuntimeError(f"GET /categories failed: HTTP {cat_status}") flat_cats: List[dict] = [] _flatten_category_tree(tree, flat_cats, parent_path_names="") counts["categories"] = cat_repo.bulk_replace(connection_id, flat_cats, now) logger.info("[baseline] Stored %d category registry entries", counts["categories"]) from db.master_taxonomy_hierarchy import reconcile_magento_channel_links_against_registry link_stats = reconcile_magento_channel_links_against_registry(sess, connection_id) counts["stale_channel_links_deactivated"] = link_stats.get("links_deactivated", 0) counts["stale_listing_targets_deactivated"] = link_stats.get("listing_targets_deactivated", 0) if counts["stale_channel_links_deactivated"] or counts["stale_listing_targets_deactivated"]: logger.info( "[baseline] Cleared stale Magento taxonomy links=%s listing_targets=%s", counts["stale_channel_links_deactivated"], counts["stale_listing_targets_deactivated"], ) # 3) Attribute sets logger.info("[baseline] Pulling attribute sets...") _delay() sets_status, sets_items, sets_err = api.list_attribute_sets() if sets_status != 200: raise RuntimeError(f"GET /products/attribute-sets/sets/list failed: HTTP {sets_status} {sets_err}") if not sets_items: raise RuntimeError("No attribute sets returned from Magento") ENTITY_TYPE_CATALOG_PRODUCT = 4 set_rows = [] for s in sets_items: if not isinstance(s, dict): continue set_id = s.get("attribute_set_id") or s.get("attribute_set_Id") name = s.get("attribute_set_name") or s.get("attribute_set_Name") or "Unknown" if set_id is None: continue # Only store catalog_product attribute sets (entity_type_id=4) # Avoids "Invalid attribute set entity type" when creating products entity_type = s.get("entity_type_id") or s.get("entityTypeId") if entity_type is not None: try: et = int(entity_type) if et != ENTITY_TYPE_CATALOG_PRODUCT: continue except (TypeError, ValueError): pass set_rows.append({"attribute_set_id": int(set_id), "attribute_set_name": str(name)}) if not set_rows: raise RuntimeError( "No catalog_product (entity_type_id=4) attribute sets in Magento response. " "Check Magento API or try without entity_type_id filter." ) counts["attribute_sets"] = attr_set_repo.bulk_replace(connection_id, set_rows, now) logger.info("[baseline] Stored %d product attribute sets (entity_type_id=4)", counts["attribute_sets"]) # 4) Determine default attribute set default_set_id: Optional[int] = None target_name = baseline_cfg.default_attribute_set_name or "Default" default_set_id = attr_set_repo.get_by_name(connection_id, target_name) if default_set_id is None: default_set_id = attr_set_repo.get_first_by_id(connection_id) logger.warning( "[baseline] Attribute set '%s' not found; using first set (id=%s)", target_name, default_set_id, ) else: logger.info("[baseline] Using default attribute set: %s (id=%s)", target_name, default_set_id) if default_set_id is None: raise RuntimeError("No attribute set found in Magento") # 5) Pull attributes for default set logger.info("[baseline] Pulling attributes for set id=%s...", default_set_id) _delay() attrs_status, set_attrs, attrs_err = api.get_attributes_for_set(default_set_id) if attrs_status != 200: raise RuntimeError(f"GET /attribute-sets/{default_set_id}/attributes failed: HTTP {attrs_status} {attrs_err}") set_attr_list = set_attrs if isinstance(set_attrs, list) else [] set_attr_count = attr_set_attrs_repo.bulk_replace_for_set( connection_id, default_set_id, set_attr_list, now ) logger.info("[baseline] Set has %d attributes in magento_attribute_set_attributes", set_attr_count) # 6) For each attribute in set: get detail + options (if select/multiselect) total_options = 0 attr_registry_count = 0 for att in set_attr_list: if not isinstance(att, dict): continue code = str(att.get("attribute_code") or att.get("attributeCode", "")).strip().lower() attr_id = att.get("attribute_id") or att.get("attributeId") if not code or attr_id is None: continue _delay() detail_status, detail = api.get_attribute_detail(code) if detail_status != 200 or not detail: logger.warning("[baseline] Attribute %s not found (HTTP %s), skipping", code, detail_status) continue attr_repo.upsert_attribute( connection_id, code, int(attr_id), frontend_label=detail.get("frontend_label"), frontend_input=detail.get("frontend_input"), backend_type=detail.get("backend_type"), is_user_defined=detail.get("is_user_defined"), fetched_at=now, ) attr_registry_count += 1 fi = str(detail.get("frontend_input", "")).strip().lower() if fi in SELECT_OR_MULTISELECT: _delay() opts_status, opts = api.get_attribute_options(code) if opts_status == 200 and opts: option_rows: List[Dict[str, Any]] = [] # Attribute-specific fallbacks for empty labels (Magento sometimes omits them) FALLBACK_BY_ATTR: Dict[str, Dict[int, str]] = { "gift_message_available": {0: "No", 1: "Yes", 2: "Use config"}, "msrp_display_actual_price_type": { 0: "Use config", 1: "On Gesture", 2: "In Cart", 3: "Before Order Confirmation" }, } fallbacks = FALLBACK_BY_ATTR.get(code, {0: "No", 1: "Yes", 2: "Use config"}) for o in opts: if not isinstance(o, dict): continue lb = o.get("label") v = o.get("value") vi_raw = o.get("value_index") val = v if (v is not None and str(v).strip() != "") else vi_raw if val is None: continue try: vi = int(val) except (ValueError, TypeError): continue label = str(lb).strip() if lb is not None else "" if not label and vi in fallbacks: label = fallbacks[vi] if label: option_rows.append({"label": label, "value_index": vi}) # Magento sometimes omits the default option (value_index 0) from the API response. # Ensure it exists for gift_message_available and msrp_display_actual_price_type. if code in FALLBACK_BY_ATTR: seen_vis = {r["value_index"] for r in option_rows} if 0 not in seen_vis and 0 in fallbacks: option_rows.append({"label": fallbacks[0], "value_index": 0}) if option_rows: attr_repo.bulk_upsert_options(connection_id, code, option_rows, now) total_options += len(option_rows) counts["attributes"] = attr_registry_count counts["options"] = total_options # Strict: if default set has >0 attributes but we inserted 0 into registry -> fail if set_attr_count > 0 and attr_registry_count == 0: raise RuntimeError( f"Default attribute set (id={default_set_id}) has {set_attr_count} attributes " "but 0 were inserted into magento_attribute_registry. Check API permissions." ) counts["default_attribute_set_id"] = default_set_id logger.info("[baseline] Stored %d attributes, %d options (default_set_id=%s)", counts["attributes"], counts["options"], default_set_id) # 6) MSI sources and stocks (persist to magento_source_registry, magento_stock_registry, magento_stock_source_links) _delay() inv_status, sources, _ = api.get_inventory_sources() if inv_status == 200 and sources: sources_list = [s for s in sources if isinstance(s, dict) and s.get("source_code")] source_count = source_repo.bulk_replace(connection_id, sources_list, now) counts["inventory_sources"] = source_count source_codes = [s.get("source_code") for s in sources_list] logger.info("[baseline] MSI sources: %d → magento_source_registry (%s)", source_count, source_codes) else: counts["inventory_sources"] = 0 logger.info("[baseline] No MSI sources (HTTP %s); magento_source_registry empty", inv_status) # 6b) MSI stocks and stock-source links _delay() stocks_status, stocks, _ = api.get_inventory_stocks() if stocks_status == 200 and stocks: stocks_list = [s for s in stocks if isinstance(s, dict) and s.get("stock_id") is not None] stock_count = stock_repo.bulk_replace(connection_id, stocks_list, now) counts["inventory_stocks"] = stock_count links: List[Dict[str, Any]] = [] for st in stocks_list: stock_id = st.get("stock_id") if stock_id is None: continue _delay() _, assigned_sources, _ = api.get_sources_assigned_to_stock(int(stock_id)) for idx, src in enumerate(assigned_sources or []): if isinstance(src, dict): sc = src.get("source_code") or src.get("code") if sc: links.append({ "stock_id": stock_id, "source_code": str(sc), "priority": src.get("priority", 100 - idx), }) links_count = stock_source_repo.bulk_replace_for_connection(connection_id, links, now) counts["inventory_stock_source_links"] = links_count logger.info( "[baseline] MSI stocks: %d, links: %d (stocks: %s)", stock_count, links_count, [(s.get("stock_id"), s.get("name")) for s in stocks_list], ) else: counts["inventory_stocks"] = 0 counts["inventory_stock_source_links"] = 0 if stocks_status != 200: logger.info("[baseline] No MSI stocks (HTTP %s)", stocks_status) if counts.get("inventory_sources", 0) == 0 and not os.getenv("MAGENTO_DEFAULT_SOURCE_CODE", "").strip(): logger.info("[baseline] Set MAGENTO_DEFAULT_SOURCE_CODE for product sync when using MSI") baseline_repo.mark_success(run_id, counts) sess.commit() logger.info("[baseline] Baseline sync completed for connection %d: %s", connection_id, counts) return {"status": "success", "run_id": run_id, "counts": counts} except Exception as e: logger.error("[baseline] Baseline sync failed for connection %d: %s", connection_id, e, exc_info=True) # IntegrityError/flush failures leave the session in PendingRollbackError; # always rollback before writing the failed run row (shared sessions especially). try: sess.rollback() except Exception: pass try: baseline_repo.mark_failed(run_id, str(e)) sess.commit() except Exception: try: sess.rollback() except Exception: pass return {"status": "failed", "run_id": run_id, "error": str(e)} class _noop_ctx: """Context manager that yields the provided object without closing.""" def __init__(self, obj): self._obj = obj def __enter__(self): return self._obj def __exit__(self, *args): pass def main() -> None: parser = argparse.ArgumentParser(description="Magento Baseline Store Sync") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--force", action="store_true", help="Ignore TTL and force refresh") args = parser.parse_args() result = run_baseline_sync(args.connection_id, force=args.force) if result["status"] == "failed": logger.error("Baseline sync failed: %s", result.get("error")) sys.exit(1) logger.info("Done: %s", result) if __name__ == "__main__": main()