""" MSI Source Backfill — assign source items to SKUs that have no sources in Magento. D5: Reuses ensure_simple_sku_inventory (same policy as main sync). One code path, one policy. For SKUs with zero source_items, assigns DEFAULT_MSI_QTY_IF_MISSING to each configured source. CLI: python -m app.jobs.magento_msi_source_backfill --connection-id X [--stock-id N] [--dry-run] Env: MAGENTO_MSI_DEFAULT_QTY_IF_MISSING=1 MAGENTO_MSI_ASSIGN_ALL_SOURCES_IF_NONE=true MAGENTO_DEFAULT_SOURCE_CODE=... (fallback if no sources in registry) MAGENTO_MSI_BACKFILL_BATCH_SIZE=50 """ from __future__ import annotations import argparse import logging import os import sys import time from typing import Any, Dict, List, Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) BATCH_SIZE = 50 # Magento often handles 50-100 source-items per request def run_msi_source_backfill( connection_id: int, *, stock_id: Optional[int] = None, dry_run: bool = False, session=None, ) -> Dict[str, Any]: """ Backfill MSI source items for SKUs that have none. Uses sources from magento_source_registry (or stock_id's linked sources). """ from db.session import get_session from db.magento_repositories import ( SqlAlchemyMagentoConnectionRepository, SqlAlchemySourceRegistryRepository, SqlAlchemyStockSourceLinksRepository, ) from db.repositories import SqlAlchemyIngestRepository from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from magento.magento_api import MagentoRestClient from settings import load_magento_sync_runtime_config runtime_cfg = load_magento_sync_runtime_config() api_delay_s = max(0, runtime_cfg.api_delay_ms) / 1000.0 batch_size = int(os.getenv("MAGENTO_MSI_BACKFILL_BATCH_SIZE", str(BATCH_SIZE))) own_session = session is None ctx = get_session() if own_session else _noop_ctx(session) with ctx as sess: conn_repo = SqlAlchemyMagentoConnectionRepository(sess) source_repo = SqlAlchemySourceRegistryRepository(sess) stock_links_repo = SqlAlchemyStockSourceLinksRepository(sess) ingest_repo = SqlAlchemyIngestRepository(sess) conn = conn_repo.get_for_sync(connection_id) if not conn: return {"status": "failed", "error": f"Connection {connection_id} not found"} # Resolve source codes: from stock_id's links, or all enabled sources, or env source_codes: List[str] = [] if stock_id: source_codes = stock_links_repo.get_sources_for_stock(connection_id, stock_id) if not source_codes: logger.warning("Stock %s has no linked sources; falling back to all sources", stock_id) if not source_codes: source_codes = source_repo.get_enabled_source_codes(connection_id) if not source_codes: source_codes = source_repo.get_all_source_codes(connection_id) if not source_codes: configured = os.getenv("MAGENTO_DEFAULT_SOURCE_CODE", "").strip() if configured: source_codes = [configured] else: return {"status": "failed", "error": "No sources in registry; run baseline sync first"} logger.info("Using sources: %s", source_codes) oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) # Detect MSI mode inv_status, sources_api, _ = api.get_inventory_sources() if inv_status != 200 or not sources_api: return {"status": "failed", "error": "MSI not enabled or sources endpoint failed"} # Get all simple SKUs from our catalog all_skus = ingest_repo.load_all_current_skus() # We need simples only; load product types from db.models import ProductSnapshot from sqlalchemy import select stmt = select(ProductSnapshot.sku, ProductSnapshot.product_type).where( ProductSnapshot.sku.in_(all_skus), ProductSnapshot.valid_to.is_(None), ) rows = sess.execute(stmt).all() simple_skus = [r.sku for r in rows if (r.product_type or "").lower() == "simple"] if not simple_skus: logger.info("No simple SKUs in catalog") return {"status": "success", "skus_checked": 0, "skus_backfilled": 0} # Chunk SKUs for source-items lookup (API accepts multiple) checked = 0 needs_backfill: List[str] = [] chunk_size = min(100, max(20, len(simple_skus))) for i in range(0, len(simple_skus), chunk_size): chunk = simple_skus[i : i + chunk_size] time.sleep(api_delay_s) status, by_sku, _ = api.get_source_items_for_skus(chunk) if status != 200: logger.warning("get_source_items_for_skus failed HTTP %s for chunk", status) continue for sku in chunk: checked += 1 items = by_sku.get(sku, []) if not items: # No source assignments - needs backfill so Magento shows in variations grid needs_backfill.append(sku) logger.info("Checked %d SKUs, %d need backfill", checked, len(needs_backfill)) if dry_run: return { "status": "dry_run", "skus_checked": checked, "skus_needing_backfill": len(needs_backfill), "sample_skus": needs_backfill[:20], "sources": source_codes, } # D5: Use same ensure_simple_sku_inventory as main sync (one policy) from settings import load_magento_msi_config msi_cfg = load_magento_msi_config() from magento.inventory_service import ensure_simple_sku_inventory source_items_posted = 0 errors: List[str] = [] for sku in needs_backfill: time.sleep(api_delay_s) result = ensure_simple_sku_inventory( api, connection_id, sku, source_codes, msi_cfg ) if result.success and result.after_source_items: source_items_posted += len(result.after_source_items) if result.mode and result.mode != "noop_preserved": logger.debug("SKU %s: %s", sku, result.mode) elif result.error: errors.append(f"{sku}: {result.error}") return { "status": "success", "skus_checked": checked, "skus_needing_backfill": len(needs_backfill), "source_items_posted": source_items_posted, "sources_used": source_codes, "errors": errors[:10] if errors else [], } class _noop_ctx: 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="MSI Source Backfill") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--stock-id", type=int, default=None, help="Limit sources to this stock's links") parser.add_argument("--dry-run", action="store_true", help="Only report what would be done") args = parser.parse_args() result = run_msi_source_backfill( args.connection_id, stock_id=args.stock_id, dry_run=args.dry_run, ) if result.get("status") == "failed": logger.error("Backfill failed: %s", result.get("error")) sys.exit(1) logger.info("Done: %s", result) if args.dry_run: print(result) if __name__ == "__main__": main()