Add helper functions for Exception
This commit is contained in:
Backend IM Bot
parent
12b83331c6
commit
a9e30e13fb
@ -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"
|
||||
|
||||
def check_duplicate_product(db: Session, name: str) -> bool:
|
||||
return {"is_valid": len(errors) == 0, "errors": errors}
|
||||
|
||||
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:
|
||||
db: Database session
|
||||
name: Product name to check
|
||||
exception: The exception that occurred
|
||||
operation: The operation being performed
|
||||
|
||||
Returns:
|
||||
bool: True if duplicate exists, False otherwise
|
||||
HTTPException with appropriate status code and detail
|
||||
"""
|
||||
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
|
||||
error_map = {
|
||||
ValidationError: (400, "Invalid product data"),
|
||||
ValueError: (400, "Invalid value provided"),
|
||||
Exception: (500, "Internal server error")
|
||||
}
|
||||
|
||||
def validate_product_update(product_data: Dict[str, Any]) -> bool:
|
||||
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:
|
||||
"""
|
||||
Validate product data for updates.
|
||||
Check if product with same name or SKU already exists.
|
||||
|
||||
Args:
|
||||
product_data: Product update data to validate
|
||||
db: Database session
|
||||
name: Product name
|
||||
sku: Product SKU
|
||||
|
||||
Returns:
|
||||
bool: True if update data is valid, False otherwise
|
||||
bool indicating if duplicate exists
|
||||
"""
|
||||
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
|
||||
|
||||
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 format_product_response(
|
||||
product: Product,
|
||||
include_timestamps: bool = False
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Formats product data for API response.
|
||||
|
||||
Args:
|
||||
product: Product model instance
|
||||
include_timestamps: Whether to include timestamps
|
||||
|
||||
Returns:
|
||||
Formatted product dictionary
|
||||
"""
|
||||
response = {
|
||||
"id": product.id,
|
||||
"name": product.name,
|
||||
"price": str(product.price),
|
||||
"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 log_product_operation(
|
||||
operation: str,
|
||||
product_id: int,
|
||||
user_id: Optional[int] = None,
|
||||
details: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Logs product operations for auditing.
|
||||
|
||||
Args:
|
||||
operation: Type of operation performed
|
||||
product_id: ID of the product
|
||||
user_id: ID of user performing operation
|
||||
details: Additional operation details
|
||||
"""
|
||||
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}"
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user