156 lines
3.8 KiB
Python
156 lines
3.8 KiB
Python
from typing import Dict, Optional, Union, Any
|
|
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
|
|
|
|
def validate_product_data(
|
|
name: str,
|
|
price: Decimal,
|
|
stock: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validates product data before creation.
|
|
|
|
Args:
|
|
name: Product name
|
|
price: Product price
|
|
stock: Stock quantity
|
|
|
|
Returns:
|
|
Dict with validation results
|
|
|
|
Raises:
|
|
ValidationError: If validation fails
|
|
"""
|
|
errors = {}
|
|
|
|
if not name or len(name) < 3:
|
|
errors["name"] = "Product name must be at least 3 characters"
|
|
|
|
if price <= Decimal(0):
|
|
errors["price"] = "Price must be greater than 0"
|
|
|
|
if stock < 0:
|
|
errors["stock"] = "Stock cannot be negative"
|
|
|
|
return {"is_valid": len(errors) == 0, "errors": errors}
|
|
|
|
def handle_product_exception(
|
|
exception: Exception,
|
|
operation: str
|
|
) -> HTTPException:
|
|
"""
|
|
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
|
|
sku: Product SKU
|
|
|
|
Returns:
|
|
bool indicating if duplicate exists
|
|
"""
|
|
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}"
|
|
) |