Add helper functions for Product

helpers/product_helpers.py

@@ -0,0 +1,104 @@
+from typing import Optional, Dict, List
+from pydantic import BaseModel, validator
+from datetime import datetime
+
+class Product(BaseModel):
+    name: str
+    description: Optional[str] = None
+    price: float
+    category: str
+    created_at: datetime = datetime.now()
+    updated_at: datetime = datetime.now()
+
+    @validator('name')
+    def validate_name(cls, value):
+        if len(value) < 3:
+            raise ValueError('Name must be at least 3 characters long')
+        return value
+
+    @validator('price')
+    def validate_price(cls, value):
+        if value < 0:
+            raise ValueError('Price cannot be negative')
+        return value
+
+def validate_product_data(data: Dict) -> Product:
+    """
+    Validate the product data dictionary and return a Product instance.
+
+    Args:
+        data (Dict): The product data dictionary.
+
+    Returns:
+        Product: A validated Product instance.
+
+    Raises:
+        ValueError: If the product data is invalid.
+    """
+    try:
+        product = Product(**data)
+    except ValueError as e:
+        raise ValueError(str(e))
+    return product
+
+def create_product(db, product_data: Dict) -> Product:
+    """
+    Create a new product in the database.
+
+    Args:
+        db: The database session.
+        product_data (Dict): The product data dictionary.
+
+    Returns:
+        Product: The created product instance.
+
+    Raises:
+        ValueError: If the product data is invalid.
+    """
+    validated_product = validate_product_data(product_data)
+    db_product = db.query(Product).filter(Product.name == validated_product.name).first()
+    if db_product:
+        raise ValueError(f'Product with name "{validated_product.name}" already exists')
+    db.add(validated_product)
+    db.commit()
+    db.refresh(validated_product)
+    return validated_product
+
+def get_products_by_category(db, category: str) -> List[Product]:
+    """
+    Get a list of products by category.
+
+    Args:
+        db: The database session.
+        category (str): The category to filter by.
+
+    Returns:
+        List[Product]: A list of products in the specified category.
+    """
+    return db.query(Product).filter(Product.category == category).all()
+
+def update_product(db, product_id: int, data: Dict) -> Product:
+    """
+    Update an existing product in the database.
+
+    Args:
+        db: The database session.
+        product_id (int): The ID of the product to update.
+        data (Dict): The updated product data dictionary.
+
+    Returns:
+        Product: The updated product instance.
+
+    Raises:
+        ValueError: If the product data is invalid or the product is not found.
+    """
+    validated_product = validate_product_data(data)
+    db_product = db.query(Product).filter(Product.id == product_id).first()
+    if not db_product:
+        raise ValueError(f'Product with ID {product_id} not found')
+    for field, value in validated_product.dict(exclude_unset=True).items():
+        setattr(db_product, field, value)
+    db_product.updated_at = datetime.now()
+    db.commit()
+    db.refresh(db_product)
+    return db_product