""" Magento Inventory Service — MSI (Multi-Source Inventory) detection and upsert. WHY THIS MATTERS: ----------------- Magento Admin "Configurations → Current Variations" grid depends on MSI source_items. GET /V1/configurable-products/{parent}/children returns all linked children, but the Admin UI only displays SKUs that have inventory_source_item rows (for quantity/salability). Without source items, variations appear "missing" even though relations are correct. We must ALWAYS upsert source items for simple products (configurable children) after product upsert, so Admin shows all variations. MSI source-items are authoritative: Magento uses them for salability. Positive qty from NetSuite/feed should be preserved when PRESERVE_EXISTING_POSITIVE_QTY is true. """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Protocol, TYPE_CHECKING if TYPE_CHECKING: from settings import MagentoMsiConfig logger = logging.getLogger(__name__) # In-memory cache: connection_id -> "msi" | "legacy" | "unknown" _MSI_MODE_CACHE: Dict[int, str] = {} _DEFAULT_SOURCE_FALLBACK = "default" @dataclass class InventoryUpsertResult: """Result of inventory upsert attempt for a single SKU.""" msi_mode_detected: bool source_items_written: List[Dict[str, Any]] = field(default_factory=list) legacy_stock_written: bool = False error: Optional[str] = None # Extended diagnostics from ensure_simple_sku_inventory mode: Optional[str] = None # created_all | patched_zero_qty | added_missing_sources | noop_preserved before_source_items: Optional[List[Dict[str, Any]]] = None after_source_items: Optional[List[Dict[str, Any]]] = None sources_added: Optional[List[str]] = None sources_updated: Optional[List[str]] = None @dataclass class EnsureSimpleSkuInventoryResult: """Structured result from ensure_simple_sku_inventory.""" success: bool mode: str # created_all | patched_zero_qty | added_missing_sources | noop_preserved before_source_items: List[Dict[str, Any]] after_source_items: List[Dict[str, Any]] sources_added: List[str] sources_updated: List[str] error: Optional[str] = None class MagentoInventoryApi(Protocol): """Minimal API surface for inventory operations.""" def get_inventory_sources(self) -> tuple[int, List[Dict[str, Any]], Optional[str]]: ... def post_source_items(self, source_items: List[Dict[str, Any]]) -> tuple[int, Optional[str]]: ... def get_source_items_for_skus(self, skus: List[str]) -> tuple[int, Dict[str, List[Dict[str, Any]]], Optional[str]]: ... def clear_msi_mode_cache(connection_id: Optional[int] = None) -> None: """ Clear MSI mode cache. If connection_id is None, clear all. Use after Magento inventory/extension changes. """ global _MSI_MODE_CACHE if connection_id is None: _MSI_MODE_CACHE.clear() else: _MSI_MODE_CACHE.pop(connection_id, None) def detect_inventory_mode(api: MagentoInventoryApi, connection_id: int) -> str: """ Detect whether Magento uses MSI (Multi-Source Inventory) or legacy stock. Caches result per connection_id. Returns "msi", "legacy", or "unknown". """ cached = _MSI_MODE_CACHE.get(connection_id) if cached is not None: return cached try: status, sources, err = api.get_inventory_sources() if status == 200 and sources is not None: _MSI_MODE_CACHE[connection_id] = "msi" return "msi" if status in (404, 501, 403) or "not found" in str(err or "").lower(): _MSI_MODE_CACHE[connection_id] = "legacy" return "legacy" except Exception as e: logger.debug("MSI detection failed for connection %s: %s", connection_id, e) _MSI_MODE_CACHE[connection_id] = "unknown" return "unknown" def resolve_source_code( api: MagentoInventoryApi, connection_id: int, configured_source: Optional[str], ) -> str: """ Resolve source_code for inventory assignment. Prefer configured (MAGENTO_DEFAULT_SOURCE_CODE), else first enabled source from API, else "default". """ if configured_source and str(configured_source).strip(): return str(configured_source).strip() try: status, sources, _ = api.get_inventory_sources() if status == 200 and sources: items = sources if isinstance(sources, list) else sources.get("items", []) for s in items: if isinstance(s, dict) and s.get("enabled", True): code = s.get("source_code") or s.get("code") if code: return str(code) except Exception: pass return _DEFAULT_SOURCE_FALLBACK def ensure_simple_sku_inventory( api: MagentoInventoryApi, connection_id: int, sku: str, source_codes: List[str], msi_config: "MagentoMsiConfig", ) -> EnsureSimpleSkuInventoryResult: """ Canonical MSI sync for simple products. Policy-driven. Only call for SIMPLE products (configurable children), not configurable parents. Policy (env-driven): - No source-items: if ASSIGN_ALL_SOURCES_IF_NONE, assign all sources with default qty, status=1 - Has source-items: - PRESERVE_EXISTING_POSITIVE_QTY: leave qty>0 unchanged - FORCE_ZERO_QTY_TO_DEFAULT: update qty<=0 to default - ASSIGN_ALL_SOURCES_IF_PARTIAL: add missing sources with default qty """ before: List[Dict[str, Any]] = [] after: List[Dict[str, Any]] = [] sources_added: List[str] = [] sources_updated: List[str] = [] mode = detect_inventory_mode(api, connection_id) if mode != "msi": return EnsureSimpleSkuInventoryResult( success=mode == "legacy", mode="noop_preserved", before_source_items=[], after_source_items=[], sources_added=[], sources_updated=[], error=None if mode == "legacy" else f"Inventory mode={mode}", ) default_qty = msi_config.default_qty_if_missing status_val = 1 # in stock # 1. Fetch current source-items from Magento try: status, by_sku, err = api.get_source_items_for_skus([sku]) if status != 200: return EnsureSimpleSkuInventoryResult( success=False, mode="noop_preserved", before_source_items=[], after_source_items=[], sources_added=[], sources_updated=[], error=err or f"HTTP {status}", ) before = list(by_sku.get(sku, [])) except Exception as e: return EnsureSimpleSkuInventoryResult( success=False, mode="noop_preserved", before_source_items=[], after_source_items=[], sources_added=[], sources_updated=[], error=str(e), ) existing_by_source: Dict[str, Dict[str, Any]] = {} for item in before: if isinstance(item, dict) and item.get("source_code"): existing_by_source[str(item["source_code"])] = item # 2. Build target source-items from policy to_post: List[Dict[str, Any]] = [] updated_set: set[str] = set() added_set: set[str] = set() if not before: # No source-items exist if msi_config.assign_all_sources_if_none and source_codes: for sc in source_codes: to_post.append({ "sku": sku, "source_code": sc, "quantity": float(default_qty), "status": status_val, }) added_set.add(sc) mode_out = "created_all" else: mode_out = "noop_preserved" else: # Source-items exist - apply policy for item in before: sc = str(item.get("source_code", "")).strip() if not sc: continue qty = item.get("quantity") try: qty_f = float(qty) if qty is not None else 0 except (TypeError, ValueError): qty_f = 0 if qty_f > 0 and msi_config.preserve_existing_positive_qty: # Leave unchanged - no-op for this source continue if qty_f <= 0 and msi_config.force_zero_qty_to_default: to_post.append({ "sku": sku, "source_code": sc, "quantity": float(default_qty), "status": status_val, }) updated_set.add(sc) if msi_config.assign_all_sources_if_partial: for sc in source_codes: if sc not in existing_by_source: to_post.append({ "sku": sku, "source_code": sc, "quantity": float(default_qty), "status": status_val, }) added_set.add(sc) sources_added = sorted(added_set) sources_updated = sorted(updated_set) if to_post: if updated_set and added_set: mode_out = "patched_zero_qty" elif updated_set: mode_out = "patched_zero_qty" else: mode_out = "added_missing_sources" else: mode_out = "noop_preserved" # 3. POST only when changes needed if not to_post: return EnsureSimpleSkuInventoryResult( success=True, mode=mode_out, before_source_items=before, after_source_items=before, sources_added=[], sources_updated=[], ) try: status, err = api.post_source_items(to_post) if status in (200, 201): # Re-fetch to get after state; Magento may not return new items immediately st2, by2, _ = api.get_source_items_for_skus([sku]) after = list(by2.get(sku, [])) if st2 == 200 and by2 else [] if not after: after = [{"source_code": si.get("source_code"), "quantity": si.get("quantity"), "status": si.get("status")} for si in to_post] return EnsureSimpleSkuInventoryResult( success=True, mode=mode_out, before_source_items=before, after_source_items=after, sources_added=sorted(added_set), sources_updated=sorted(updated_set), ) return EnsureSimpleSkuInventoryResult( success=False, mode=mode_out, before_source_items=before, after_source_items=before, sources_added=[], sources_updated=[], error=err or f"HTTP {status}", ) except Exception as e: return EnsureSimpleSkuInventoryResult( success=False, mode=mode_out if to_post else "noop_preserved", before_source_items=before, after_source_items=before, sources_added=[], sources_updated=[], error=str(e), ) def upsert_source_items_for_simple( api: MagentoInventoryApi, sku: str, row: Dict[str, Any], *, connection_id: int = 0, configured_source: Optional[str] = None, ) -> InventoryUpsertResult: """ Legacy MSI upsert - single source, qty from row (0 if missing). DEPRECATED: Use ensure_simple_sku_inventory with source_codes and MagentoMsiConfig instead. Kept for backward compatibility; delegates to ensure_simple_sku_inventory when config loaded. """ try: from settings import load_magento_msi_config msi_cfg = load_magento_msi_config() source_code = resolve_source_code(api, connection_id, configured_source) result = ensure_simple_sku_inventory( api, connection_id, sku, [source_code], msi_cfg ) return InventoryUpsertResult( msi_mode_detected=detect_inventory_mode(api, connection_id) == "msi", source_items_written=[ {"source_code": si.get("source_code"), "quantity": si.get("quantity"), "status": si.get("status")} for si in (result.after_source_items or []) ], error=result.error, mode=result.mode, before_source_items=result.before_source_items, after_source_items=result.after_source_items, sources_added=result.sources_added, sources_updated=result.sources_updated, ) except ImportError: pass mode = detect_inventory_mode(api, connection_id) if mode != "msi": return InventoryUpsertResult( msi_mode_detected=False, error=None if mode == "legacy" else f"Inventory mode={mode}", ) source_code = resolve_source_code(api, connection_id, configured_source) qty = 0.0 try: q = row.get("qty") or row.get("quantity") if q is not None: qty = float(q) except (TypeError, ValueError): pass qty = max(0, qty) in_stock = 1 if qty > 0 else 0 source_items = [{"sku": sku, "source_code": source_code, "quantity": float(qty), "status": in_stock}] written = [{"source_code": source_code, "quantity": qty, "status": in_stock}] try: status, err = api.post_source_items(source_items) if status in (200, 201): return InventoryUpsertResult(msi_mode_detected=True, source_items_written=written) return InventoryUpsertResult(msi_mode_detected=True, source_items_written=[], error=err or f"HTTP {status}") except Exception as e: return InventoryUpsertResult(msi_mode_detected=True, error=str(e))