90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
from typing import Dict
|
|
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from db.models import Product, Inventory
|
|
from schemas.inventory import UpdateInventoryResponse
|
|
from 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) |