# shared/clients/inventory_client.py """ Inventory Service Client - Inter-service communication Handles communication with the inventory service for all other services """ import structlog from typing import Dict, Any, List, Optional, Union from uuid import UUID from shared.clients.base_service_client import BaseServiceClient from shared.config.base import BaseServiceSettings logger = structlog.get_logger() class InventoryServiceClient(BaseServiceClient): """Client for communicating with the inventory service via gateway""" def __init__(self, config: BaseServiceSettings, calling_service_name: str = "unknown"): super().__init__(calling_service_name, config) def get_service_base_path(self) -> str: """Return the base path for inventory service APIs""" return "/api/v1" # ================================================================ # INGREDIENT MANAGEMENT # ================================================================ async def get_ingredient_by_id(self, ingredient_id: UUID, tenant_id: str) -> Optional[Dict[str, Any]]: """Get ingredient details by ID""" try: result = await self.get(f"inventory/ingredients/{ingredient_id}", tenant_id=tenant_id) if result: logger.info("Retrieved ingredient from inventory service", ingredient_id=ingredient_id, tenant_id=tenant_id) return result except Exception as e: logger.error("Error fetching ingredient by ID", error=str(e), ingredient_id=ingredient_id, tenant_id=tenant_id) return None async def search_ingredients( self, tenant_id: str, search: Optional[str] = None, category: Optional[str] = None, is_active: Optional[bool] = None, skip: int = 0, limit: int = 100 ) -> List[Dict[str, Any]]: """Search ingredients with filters""" try: params = { "skip": skip, "limit": limit } if search: params["search"] = search if category: params["category"] = category if is_active is not None: params["is_active"] = is_active result = await self.get("inventory/ingredients", tenant_id=tenant_id, params=params) ingredients = result if isinstance(result, list) else [] logger.info("Searched ingredients in inventory service", search_term=search, count=len(ingredients), tenant_id=tenant_id) return ingredients except Exception as e: logger.error("Error searching ingredients", error=str(e), search=search, tenant_id=tenant_id) return [] async def get_all_ingredients(self, tenant_id: str, is_active: Optional[bool] = True) -> List[Dict[str, Any]]: """Get all ingredients for a tenant (paginated)""" try: params = {} if is_active is not None: params["is_active"] = is_active ingredients = await self.get_paginated("inventory/ingredients", tenant_id=tenant_id, params=params) logger.info("Retrieved all ingredients from inventory service", count=len(ingredients), tenant_id=tenant_id) return ingredients except Exception as e: logger.error("Error fetching all ingredients", error=str(e), tenant_id=tenant_id) return [] async def count_ingredients(self, tenant_id: str, is_active: Optional[bool] = True) -> int: """Get count of ingredients for a tenant""" try: params = {} if is_active is not None: params["is_active"] = is_active result = await self.get("inventory/ingredients/count", tenant_id=tenant_id, params=params) count = result.get("ingredient_count", 0) if isinstance(result, dict) else 0 logger.info("Retrieved ingredient count from inventory service", count=count, tenant_id=tenant_id) return count except Exception as e: logger.error("Error fetching ingredient count", error=str(e), tenant_id=tenant_id) return 0 async def create_ingredient(self, ingredient_data: Dict[str, Any], tenant_id: str) -> Optional[Dict[str, Any]]: """Create a new ingredient""" try: result = await self.post("inventory/ingredients", data=ingredient_data, tenant_id=tenant_id) if result: logger.info("Created ingredient in inventory service", ingredient_name=ingredient_data.get('name'), tenant_id=tenant_id) return result except Exception as e: logger.error("Error creating ingredient", error=str(e), ingredient_data=ingredient_data, tenant_id=tenant_id) return None async def update_ingredient( self, ingredient_id: UUID, ingredient_data: Dict[str, Any], tenant_id: str ) -> Optional[Dict[str, Any]]: """Update an existing ingredient""" try: result = await self.put(f"inventory/ingredients/{ingredient_id}", data=ingredient_data, tenant_id=tenant_id) if result: logger.info("Updated ingredient in inventory service", ingredient_id=ingredient_id, tenant_id=tenant_id) return result except Exception as e: logger.error("Error updating ingredient", error=str(e), ingredient_id=ingredient_id, tenant_id=tenant_id) return None async def delete_ingredient(self, ingredient_id: UUID, tenant_id: str) -> bool: """Delete (deactivate) an ingredient""" try: result = await self.delete(f"inventory/ingredients/{ingredient_id}", tenant_id=tenant_id) success = result is not None if success: logger.info("Deleted ingredient in inventory service", ingredient_id=ingredient_id, tenant_id=tenant_id) return success except Exception as e: logger.error("Error deleting ingredient", error=str(e), ingredient_id=ingredient_id, tenant_id=tenant_id) return False async def get_ingredient_stock( self, ingredient_id: UUID, tenant_id: str, include_unavailable: bool = False ) -> List[Dict[str, Any]]: """Get stock entries for an ingredient""" try: params = {} if include_unavailable: params["include_unavailable"] = include_unavailable result = await self.get(f"inventory/ingredients/{ingredient_id}/stock", tenant_id=tenant_id, params=params) stock_entries = result if isinstance(result, list) else [] logger.info("Retrieved ingredient stock from inventory service", ingredient_id=ingredient_id, stock_count=len(stock_entries), tenant_id=tenant_id) return stock_entries except Exception as e: logger.error("Error fetching ingredient stock", error=str(e), ingredient_id=ingredient_id, tenant_id=tenant_id) return [] # ================================================================ # STOCK MANAGEMENT # ================================================================ async def get_stock_levels(self, tenant_id: str, ingredient_ids: Optional[List[UUID]] = None) -> List[Dict[str, Any]]: """Get current stock levels""" try: params = {} if ingredient_ids: params["ingredient_ids"] = [str(id) for id in ingredient_ids] result = await self.get("inventory/stock", tenant_id=tenant_id, params=params) stock_levels = result if isinstance(result, list) else [] logger.info("Retrieved stock levels from inventory service", count=len(stock_levels), tenant_id=tenant_id) return stock_levels except Exception as e: logger.error("Error fetching stock levels", error=str(e), tenant_id=tenant_id) return [] async def get_low_stock_alerts(self, tenant_id: str) -> List[Dict[str, Any]]: """Get low stock alerts""" try: result = await self.get("inventory/alerts", tenant_id=tenant_id, params={"type": "low_stock"}) alerts = result if isinstance(result, list) else [] logger.info("Retrieved low stock alerts from inventory service", count=len(alerts), tenant_id=tenant_id) return alerts except Exception as e: logger.error("Error fetching low stock alerts", error=str(e), tenant_id=tenant_id) return [] async def consume_stock( self, consumption_data: Dict[str, Any], tenant_id: str ) -> Optional[Dict[str, Any]]: """Record stock consumption""" try: result = await self.post("inventory/operations/consume-stock", data=consumption_data, tenant_id=tenant_id) if result: logger.info("Recorded stock consumption", tenant_id=tenant_id) return result except Exception as e: logger.error("Error recording stock consumption", error=str(e), tenant_id=tenant_id) return None async def receive_stock( self, receipt_data: Dict[str, Any], tenant_id: str ) -> Optional[Dict[str, Any]]: """Record stock receipt""" try: result = await self.post("inventory/operations/receive-stock", data=receipt_data, tenant_id=tenant_id) if result: logger.info("Recorded stock receipt", tenant_id=tenant_id) return result except Exception as e: logger.error("Error recording stock receipt", error=str(e), tenant_id=tenant_id) return None # ================================================================ # PRODUCT CLASSIFICATION (for onboarding) # ================================================================ async def classify_product( self, product_name: str, sales_volume: Optional[float], tenant_id: str ) -> Optional[Dict[str, Any]]: """Classify a single product for inventory creation""" try: classification_data = { "product_name": product_name, "sales_volume": sales_volume } result = await self.post("inventory/operations/classify-product", data=classification_data, tenant_id=tenant_id) if result: logger.info("Classified product", product=product_name, classification=result.get('product_type'), confidence=result.get('confidence_score'), tenant_id=tenant_id) return result except Exception as e: logger.error("Error classifying product", error=str(e), product=product_name, tenant_id=tenant_id) return None async def classify_products_batch( self, products: List[Dict[str, Any]], tenant_id: str ) -> Optional[Dict[str, Any]]: """Classify multiple products for onboarding automation""" try: classification_data = { "products": products } result = await self.post("inventory/operations/classify-products-batch", data=classification_data, tenant_id=tenant_id) if result: suggestions = result.get('suggestions', []) business_model = result.get('business_model_analysis', {}).get('model', 'unknown') logger.info("Batch classification complete", total_products=len(suggestions), business_model=business_model, tenant_id=tenant_id) return result except Exception as e: logger.error("Error in batch classification", error=str(e), products_count=len(products), tenant_id=tenant_id) return None async def resolve_or_create_products_batch( self, products: List[Dict[str, Any]], tenant_id: str ) -> Optional[Dict[str, Any]]: """Resolve or create multiple products in a single batch operation""" try: batch_data = { "products": products } result = await self.post("inventory/operations/resolve-or-create-products-batch", data=batch_data, tenant_id=tenant_id) if result: created = result.get('created_count', 0) resolved = result.get('resolved_count', 0) failed = result.get('failed_count', 0) logger.info("Batch product resolution complete", created=created, resolved=resolved, failed=failed, total=len(products), tenant_id=tenant_id) return result except Exception as e: logger.error("Error in batch product resolution", error=str(e), products_count=len(products), tenant_id=tenant_id) return None # ================================================================ # DASHBOARD AND ANALYTICS # ================================================================ async def get_inventory_dashboard(self, tenant_id: str) -> Optional[Dict[str, Any]]: """Get inventory dashboard data""" try: result = await self.get("inventory/dashboard/overview", tenant_id=tenant_id) if result: logger.info("Retrieved inventory dashboard data", tenant_id=tenant_id) return result except Exception as e: logger.error("Error fetching inventory dashboard", error=str(e), tenant_id=tenant_id) return None async def get_inventory_summary(self, tenant_id: str) -> Optional[Dict[str, Any]]: """Get inventory summary statistics""" try: result = await self.get("inventory/dashboard/summary", tenant_id=tenant_id) if result: logger.info("Retrieved inventory summary", tenant_id=tenant_id) return result except Exception as e: logger.error("Error fetching inventory summary", error=str(e), tenant_id=tenant_id) return None # ================================================================ # PRODUCT TRANSFORMATION # ================================================================ async def create_transformation( self, transformation_data: Dict[str, Any], tenant_id: str ) -> Optional[Dict[str, Any]]: """Create a product transformation (e.g., par-baked to fully baked)""" try: result = await self.post("inventory/transformations", data=transformation_data, tenant_id=tenant_id) if result: logger.info("Created product transformation", transformation_reference=result.get('transformation_reference'), source_stage=transformation_data.get('source_stage'), target_stage=transformation_data.get('target_stage'), tenant_id=tenant_id) return result except Exception as e: logger.error("Error creating transformation", error=str(e), transformation_data=transformation_data, tenant_id=tenant_id) return None async def create_par_bake_transformation( self, source_ingredient_id: Union[str, UUID], target_ingredient_id: Union[str, UUID], quantity: float, tenant_id: str, target_batch_number: Optional[str] = None, expiration_hours: int = 24, notes: Optional[str] = None ) -> Optional[Dict[str, Any]]: """Convenience method for par-baked to fresh transformation""" try: params = { "source_ingredient_id": str(source_ingredient_id), "target_ingredient_id": str(target_ingredient_id), "quantity": quantity, "expiration_hours": expiration_hours } if target_batch_number: params["target_batch_number"] = target_batch_number if notes: params["notes"] = notes result = await self.post("inventory/transformations/par-bake-to-fresh", params=params, tenant_id=tenant_id) if result: logger.info("Created par-bake transformation", transformation_id=result.get('transformation_id'), quantity=quantity, tenant_id=tenant_id) return result except Exception as e: logger.error("Error creating par-bake transformation", error=str(e), source_ingredient_id=source_ingredient_id, target_ingredient_id=target_ingredient_id, tenant_id=tenant_id) return None async def get_transformations( self, tenant_id: str, ingredient_id: Optional[Union[str, UUID]] = None, source_stage: Optional[str] = None, target_stage: Optional[str] = None, days_back: Optional[int] = None, skip: int = 0, limit: int = 100 ) -> List[Dict[str, Any]]: """Get product transformations with filtering""" try: params = { "skip": skip, "limit": limit } if ingredient_id: params["ingredient_id"] = str(ingredient_id) if source_stage: params["source_stage"] = source_stage if target_stage: params["target_stage"] = target_stage if days_back: params["days_back"] = days_back result = await self.get("inventory/transformations", tenant_id=tenant_id, params=params) transformations = result if isinstance(result, list) else [] logger.info("Retrieved transformations from inventory service", count=len(transformations), tenant_id=tenant_id) return transformations except Exception as e: logger.error("Error fetching transformations", error=str(e), tenant_id=tenant_id) return [] async def get_transformation_by_id( self, transformation_id: Union[str, UUID], tenant_id: str ) -> Optional[Dict[str, Any]]: """Get specific transformation by ID""" try: result = await self.get(f"inventory/transformations/{transformation_id}", tenant_id=tenant_id) if result: logger.info("Retrieved transformation by ID", transformation_id=transformation_id, tenant_id=tenant_id) return result except Exception as e: logger.error("Error fetching transformation by ID", error=str(e), transformation_id=transformation_id, tenant_id=tenant_id) return None async def get_transformation_summary( self, tenant_id: str, days_back: int = 30 ) -> Optional[Dict[str, Any]]: """Get transformation summary for dashboard""" try: params = {"days_back": days_back} result = await self.get("inventory/dashboard/transformations-summary", tenant_id=tenant_id, params=params) if result: logger.info("Retrieved transformation summary", days_back=days_back, tenant_id=tenant_id) return result except Exception as e: logger.error("Error fetching transformation summary", error=str(e), tenant_id=tenant_id) return None # ================================================================ # BATCH OPERATIONS (NEW - for Orchestrator optimization) # ================================================================ async def get_ingredients_batch( self, tenant_id: str, ingredient_ids: List[UUID] ) -> Dict[str, Any]: """ Fetch multiple ingredients in a single request. This method reduces N API calls to 1, significantly improving performance when fetching data for multiple ingredients. Args: tenant_id: Tenant ID ingredient_ids: List of ingredient IDs to fetch Returns: Dict with 'ingredients', 'found_count', and 'missing_ids' """ try: if not ingredient_ids: return { 'ingredients': [], 'found_count': 0, 'missing_ids': [] } # Convert UUIDs to strings for JSON serialization ids_str = [str(id) for id in ingredient_ids] result = await self.post( "inventory/operations/ingredients/batch", data={"ingredient_ids": ids_str}, tenant_id=tenant_id ) if result: logger.info( "Retrieved ingredients in batch", requested=len(ingredient_ids), found=result.get('found_count', 0), tenant_id=tenant_id ) return result or {'ingredients': [], 'found_count': 0, 'missing_ids': ids_str} except Exception as e: logger.error( "Error fetching ingredients in batch", error=str(e), count=len(ingredient_ids), tenant_id=tenant_id ) return {'ingredients': [], 'found_count': 0, 'missing_ids': [str(id) for id in ingredient_ids]} async def get_stock_levels_batch( self, tenant_id: str, ingredient_ids: List[UUID] ) -> Dict[str, float]: """ Fetch stock levels for multiple ingredients in a single request. Args: tenant_id: Tenant ID ingredient_ids: List of ingredient IDs Returns: Dict mapping ingredient_id (str) to stock level (float) """ try: if not ingredient_ids: return {} # Convert UUIDs to strings for JSON serialization ids_str = [str(id) for id in ingredient_ids] result = await self.post( "inventory/operations/stock-levels/batch", data={"ingredient_ids": ids_str}, tenant_id=tenant_id ) stock_levels = result.get('stock_levels', {}) if result else {} logger.info( "Retrieved stock levels in batch", requested=len(ingredient_ids), found=len(stock_levels), tenant_id=tenant_id ) return stock_levels except Exception as e: logger.error( "Error fetching stock levels in batch", error=str(e), count=len(ingredient_ids), tenant_id=tenant_id ) return {} # ================================================================ # ML INSIGHTS: Safety Stock Optimization # ================================================================ async def trigger_safety_stock_optimization( self, tenant_id: str, product_ids: Optional[List[str]] = None, lookback_days: int = 90, min_history_days: int = 30 ) -> Optional[Dict[str, Any]]: """ Trigger safety stock optimization for inventory products. Args: tenant_id: Tenant UUID product_ids: Specific product IDs to optimize. If None, optimizes all products lookback_days: Days of historical demand to analyze (30-365) min_history_days: Minimum days of history required (7-180) Returns: Dict with optimization results including insights posted """ try: data = { "product_ids": product_ids, "lookback_days": lookback_days, "min_history_days": min_history_days } result = await self.post("inventory/ml/insights/optimize-safety-stock", data=data, tenant_id=tenant_id) if result: logger.info("Triggered safety stock optimization", products_optimized=result.get('products_optimized', 0), insights_posted=result.get('total_insights_posted', 0), tenant_id=tenant_id) return result except Exception as e: logger.error("Error triggering safety stock optimization", error=str(e), tenant_id=tenant_id) return None # ================================================================ # UTILITY METHODS # ================================================================ async def health_check(self) -> bool: """Check if inventory service is healthy""" try: result = await self.get("../health") # Health endpoint is not tenant-scoped return result is not None except Exception as e: logger.error("Inventory service health check failed", error=str(e)) return False # Factory function for dependency injection def create_inventory_client(config: BaseServiceSettings) -> InventoryServiceClient: """Create inventory service client instance""" return InventoryServiceClient(config) # Convenience function for quick access (requires config to be passed) async def get_inventory_client(config: BaseServiceSettings) -> InventoryServiceClient: """Get inventory service client instance""" return create_inventory_client(config)