""" @service InventoryService (Backend/app/services/InventoryService.py) @purpose Service layer helper functions to calculate available stock, physical stock, and reserved stock dynamically by querying the append-only inventory_ledger. """ from sqlalchemy.orm import Session from sqlalchemy import func from typing import Dict, Iterable, Optional from app.models.InventoryLedgerModel import InventoryLedger from app.models.OrderModel import Order, OrderItem import ulid def get_available_stock(variant_id: str, db: Session) -> int: """ Calculate Available Stock by summing all append-only qty records in the ledger for the variant. """ qty_sum = db.query(func.sum(InventoryLedger.qty)).filter(InventoryLedger.variant_id == variant_id).scalar() return int(qty_sum) if qty_sum is not None else 0 def get_available_stock_map(variant_ids: Iterable[str], db: Session) -> Dict[str, int]: """ Batch stock lookup for a page of variants (avoids N+1 at 500k+ catalog scale). """ ids = [vid for vid in set(variant_ids) if vid] if not ids: return {} rows = ( db.query(InventoryLedger.variant_id, func.coalesce(func.sum(InventoryLedger.qty), 0)) .filter(InventoryLedger.variant_id.in_(ids)) .group_by(InventoryLedger.variant_id) .all() ) stock_map = {vid: 0 for vid in ids} for variant_id, qty in rows: stock_map[variant_id] = int(qty or 0) return stock_map def get_reserved_stock(variant_id: str, db: Session) -> int: """ Calculate Reserved Stock by summing variant quantities in active orders that have not been confirmed/cancelled yet. Active order statuses: ORDER_CREATED, PAYMENT_PENDING. """ reserved_qty = ( db.query(func.sum(OrderItem.quantity)) .join(Order, Order.order_id == OrderItem.order_id) .filter( OrderItem.variant_id == variant_id, Order.status.in_(["ORDER_CREATED", "PAYMENT_PENDING"]) ) .scalar() ) return int(reserved_qty) if reserved_qty is not None else 0 def get_stock_metrics(variant_id: str, db: Session) -> dict: """ Get all stock buckets for a variant: physical, reserved, available, pending, and confirmed. """ available = get_available_stock(variant_id, db) pending_conf = get_pending_confirmation_units(variant_id, db) confirmed = get_confirmed_units(variant_id, db) # Physical stock represents available units plus those currently held on confirmed status physical = available + confirmed # Mock/simulated damaged and pos allocation metrics matching schema stubs damaged = 0 pos_allocated = 0 return { "variant_id": variant_id, "physical_stock": physical, "reserved_stock": pending_conf, "pending_confirmation_units": pending_conf, "confirmed_units": confirmed, "pos_allocated_stock": pos_allocated, "damaged_stock": damaged, "available_stock": available } def record_ledger_entry( variant_id: str, event_type: str, qty: int, reference_id: str, db: Session, notes: str = None, *, commit: bool = True, ) -> InventoryLedger: """ Write a new append-only entry to the inventory ledger. Set commit=False when the caller manages a larger transaction (e.g. checkout). """ entry = InventoryLedger( ledger_id=str(ulid.ULID()), variant_id=variant_id, event_type=event_type, qty=qty, reference_id=reference_id, notes=notes ) db.add(entry) if commit: db.commit() db.refresh(entry) else: db.flush() return entry def apply_stock_target( variant_id: str, target_qty: Optional[int], db: Session, *, notes: str = "Product catalog stock set", commit: bool = False, ) -> Optional[InventoryLedger]: """ Set sellable stock to an absolute number (Sassynest-style stock_quantity). Writes a RECEIPT (positive) or ADJUSTMENT (negative) delta against the ledger. """ if target_qty is None: return None target_qty = max(0, int(target_qty)) current = get_available_stock(variant_id, db) delta = target_qty - current if delta == 0: return None event_type = "RECEIPT" if delta > 0 else "ADJUSTMENT" return record_ledger_entry( variant_id=variant_id, event_type=event_type, qty=delta, reference_id="CATALOG_STOCK", db=db, notes=notes, commit=commit, ) from datetime import datetime, timedelta def get_pending_confirmation_units_map(variant_ids: Iterable[str], db: Session) -> Dict[str, int]: """ Calculate Pending Confirmation Units dynamically for a list of variants. Unconfirmed orders are those with status in ORDER_CREATED or PAYMENT_PENDING. """ ids = [vid for vid in set(variant_ids) if vid] if not ids: return {} rows = ( db.query(OrderItem.variant_id, func.sum(OrderItem.quantity)) .join(Order, Order.order_id == OrderItem.order_id) .filter( OrderItem.variant_id.in_(ids), Order.status.in_(["ORDER_CREATED", "PAYMENT_PENDING"]) ) .group_by(OrderItem.variant_id) .all() ) result = {vid: 0 for vid in ids} for variant_id, qty in rows: result[variant_id] = int(qty or 0) return result def get_confirmed_units_map(variant_ids: Iterable[str], db: Session) -> Dict[str, int]: """ Calculate Confirmed Units dynamically for a list of variants. Confirmed orders are in status ORDER_CONFIRMED, PROCESSING, PACKED, SHIPPED. Delivered orders are held in confirmed units for a 6-day return window. """ ids = [vid for vid in set(variant_ids) if vid] if not ids: return {} six_days_ago = datetime.utcnow() - timedelta(days=6) rows = ( db.query(OrderItem.variant_id, func.sum(OrderItem.quantity)) .join(Order, Order.order_id == OrderItem.order_id) .filter( OrderItem.variant_id.in_(ids), ( Order.status.in_(["ORDER_CONFIRMED", "PROCESSING", "PACKED", "SHIPPED"]) | ((Order.status == "DELIVERED") & (Order.updated_at >= six_days_ago)) ) ) .group_by(OrderItem.variant_id) .all() ) result = {vid: 0 for vid in ids} for variant_id, qty in rows: result[variant_id] = int(qty or 0) return result def get_pending_confirmation_units(variant_id: str, db: Session) -> int: res = get_pending_confirmation_units_map([variant_id], db) return res.get(variant_id, 0) def get_confirmed_units(variant_id: str, db: Session) -> int: res = get_confirmed_units_map([variant_id], db) return res.get(variant_id, 0)