diff --git a/helpers/exception_helpers.py b/helpers/exception_helpers.py index 5f18a68..a801866 100644 --- a/helpers/exception_helpers.py +++ b/helpers/exception_helpers.py @@ -1,117 +1,156 @@ from typing import Dict, Optional, Union, Any -from decimal import Decimal -from sqlalchemy.orm import Session from fastapi import HTTPException +from pydantic import BaseModel, ValidationError +from decimal import Decimal +import logging +from datetime import datetime +from sqlalchemy.orm import Session from models.product import Product -from schemas.product import ProductCreate -def validate_product_data(product_data: ProductCreate) -> bool: +def validate_product_data( + name: str, + price: Decimal, + stock: int +) -> Dict[str, Any]: """ - Validate product data before creation. + Validates product data before creation. Args: - product_data: Product data to validate + name: Product name + price: Product price + stock: Stock quantity Returns: - bool: True if data is valid, False otherwise + Dict with validation results + + Raises: + ValidationError: If validation fails """ - if not product_data.name or len(product_data.name.strip()) == 0: - return False + errors = {} - if not product_data.price or product_data.price <= Decimal('0.00'): - return False + if not name or len(name) < 3: + errors["name"] = "Product name must be at least 3 characters" - if product_data.quantity and product_data.quantity < 0: - return False + if price <= Decimal(0): + errors["price"] = "Price must be greater than 0" - return True + if stock < 0: + errors["stock"] = "Stock cannot be negative" + + return {"is_valid": len(errors) == 0, "errors": errors} -def check_duplicate_product(db: Session, name: str) -> bool: +def handle_product_exception( + exception: Exception, + operation: str +) -> HTTPException: """ - Check if a product with the same name already exists. + Handles product related exceptions and returns appropriate HTTP exceptions. + + Args: + exception: The exception that occurred + operation: The operation being performed + + Returns: + HTTPException with appropriate status code and detail + """ + error_map = { + ValidationError: (400, "Invalid product data"), + ValueError: (400, "Invalid value provided"), + Exception: (500, "Internal server error") + } + + status_code, detail = error_map.get( + type(exception), + (500, "Internal server error") + ) + + logging.error( + f"Error during {operation}: {str(exception)}" + ) + + return HTTPException( + status_code=status_code, + detail=detail + ) + +def check_duplicate_product( + db: Session, + name: str, + sku: Optional[str] = None +) -> bool: + """ + Check if product with same name or SKU already exists. Args: db: Database session - name: Product name to check + name: Product name + sku: Product SKU Returns: - bool: True if duplicate exists, False otherwise + bool indicating if duplicate exists """ - existing_product = db.query(Product).filter(Product.name == name).first() - return bool(existing_product) + query = db.query(Product) + if query.filter(Product.name == name).first(): + return True + + if sku and query.filter(Product.sku == sku).first(): + return True + + return False -def create_product_safely(db: Session, product_data: ProductCreate) -> Union[Product, Dict[str, str]]: +def format_product_response( + product: Product, + include_timestamps: bool = False +) -> Dict[str, Any]: """ - Create a new product with validation and error handling. + Formats product data for API response. Args: - db: Database session - product_data: Product data for creation + product: Product model instance + include_timestamps: Whether to include timestamps Returns: - Product object if created successfully, error dict otherwise + Formatted product dictionary """ - 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 { + response = { "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 + "stock": product.stock, + "sku": product.sku } + + if include_timestamps: + response.update({ + "created_at": product.created_at.isoformat(), + "updated_at": product.updated_at.isoformat() if product.updated_at else None + }) + + return response -def validate_product_update(product_data: Dict[str, Any]) -> bool: +def log_product_operation( + operation: str, + product_id: int, + user_id: Optional[int] = None, + details: Optional[Dict] = None +) -> None: """ - Validate product data for updates. + Logs product operations for auditing. Args: - product_data: Product update data to validate - - Returns: - bool: True if update data is valid, False otherwise + operation: Type of operation performed + product_id: ID of the product + user_id: ID of user performing operation + details: Additional operation details """ - 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 + log_entry = { + "timestamp": datetime.utcnow().isoformat(), + "operation": operation, + "product_id": product_id, + "user_id": user_id, + "details": details or {} + } + + logging.info( + f"Product Operation: {log_entry}" + ) \ No newline at end of file