104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
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 |