diff --git a/helpers/exception_helpers.py b/helpers/exception_helpers.py new file mode 100644 index 0000000..5f18a68 --- /dev/null +++ b/helpers/exception_helpers.py @@ -0,0 +1,117 @@ +from typing import Dict, Optional, Union, Any +from decimal import Decimal +from sqlalchemy.orm import Session +from fastapi import HTTPException +from models.product import Product +from schemas.product import ProductCreate + +def validate_product_data(product_data: ProductCreate) -> bool: + """ + Validate product data before creation. + + Args: + product_data: Product data to validate + + Returns: + bool: True if data is valid, False otherwise + """ + if not product_data.name or len(product_data.name.strip()) == 0: + return False + + if not product_data.price or product_data.price <= Decimal('0.00'): + return False + + if product_data.quantity and product_data.quantity < 0: + return False + + return True + +def check_duplicate_product(db: Session, name: str) -> bool: + """ + Check if a product with the same name already exists. + + Args: + db: Database session + name: Product name to check + + Returns: + bool: True if duplicate exists, False otherwise + """ + existing_product = db.query(Product).filter(Product.name == name).first() + return bool(existing_product) + +def create_product_safely(db: Session, product_data: ProductCreate) -> Union[Product, Dict[str, str]]: + """ + Create a new product with validation and error handling. + + Args: + db: Database session + product_data: Product data for creation + + Returns: + Product object if created successfully, error dict otherwise + """ + if not validate_product_data(product_data): + raise HTTPException(status_code=400, detail="Invalid product data") + + if check_duplicate_product(db, product_data.name): + raise HTTPException(status_code=400, detail="Product with this name already exists") + + try: + db_product = Product( + name=product_data.name, + description=product_data.description, + price=product_data.price, + quantity=product_data.quantity, + category_id=product_data.category_id + ) + + db.add(db_product) + db.commit() + db.refresh(db_product) + return db_product + + except Exception as e: + db.rollback() + raise HTTPException(status_code=500, detail=f"Failed to create product: {str(e)}") + +def format_product_response(product: Product) -> Dict[str, Any]: + """ + Format product data for API response. + + Args: + product: Product object to format + + Returns: + Dict containing formatted product data + """ + return { + "id": product.id, + "name": product.name, + "description": product.description, + "price": str(product.price), + "quantity": product.quantity, + "category_id": product.category_id, + "created_at": product.created_at.isoformat() if product.created_at else None + } + +def validate_product_update(product_data: Dict[str, Any]) -> bool: + """ + Validate product data for updates. + + Args: + product_data: Product update data to validate + + Returns: + bool: True if update data is valid, False otherwise + """ + if "price" in product_data and product_data["price"] <= Decimal('0.00'): + return False + + if "quantity" in product_data and product_data["quantity"] < 0: + return False + + if "name" in product_data and len(product_data["name"].strip()) == 0: + return False + + return True \ No newline at end of file