from typing import Dict from sqlalchemy.orm import Session from sqlalchemy.exc import SQLAlchemyError from app.db.models import Product, Inventory from app.schemas.inventory import UpdateInventoryResponse from app.db.utils import db_transaction class InventoryService: """Service class for managing product inventory operations.""" def __init__(self, db: Session) -> None: """ Initialize the InventoryService. Args: db: SQLAlchemy database session """ self.db = db def add_inventory(self, product: Product, quantity: int) -> Inventory: """ Add or update inventory for a product. Args: product: Product model instance quantity: Quantity to add to inventory Returns: Updated Inventory model instance """ inventory = self.db.query(Inventory).filter( Inventory.product_id == product.id ).first() if inventory is None: inventory = Inventory( product_id=product.id, quantity=quantity ) self.db.add(inventory) else: inventory.quantity += quantity return inventory def process_staged_products( self, product_data: Dict[Product, int] ) -> UpdateInventoryResponse: """ Process multiple products and update their inventory. Args: product_data: Dictionary mapping Products to their quantities Returns: Response indicating success status """ try: with db_transaction(self.db): for product, quantity in product_data.items(): self.add_inventory(product, quantity) return UpdateInventoryResponse(success=True) except SQLAlchemyError: return UpdateInventoryResponse(success=False) def add_sealed_box_to_inventory( self, product: Product, quantity: int ) -> UpdateInventoryResponse: """ Add sealed box inventory for a single product. Args: product: Product model instance quantity: Quantity to add to inventory Returns: Response indicating success status """ try: with db_transaction(self.db): self.add_inventory(product, quantity) return UpdateInventoryResponse(success=True) except SQLAlchemyError: return UpdateInventoryResponse(success=False)