ecommerceauthenticationandinventoryapi-df76gf/app/crud/product.py

from typing import Any, Dict, List, Optional, Union

from sqlalchemy.orm import Session

from app.models.product import Category, Product
from app.schemas.product import (
    CategoryCreate,
    CategoryUpdate,
    ProductCreate,
    ProductUpdate,
)


# Product CRUD operations
def get_product_by_id(db: Session, product_id: str) -> Optional[Product]:
    """
    Get a product by ID.
    
    Args:
        db: Database session
        product_id: Product ID
        
    Returns:
        Optional[Product]: Product if found, None otherwise
    """
    return db.query(Product).filter(Product.id == product_id).first()


def get_product_by_sku(db: Session, sku: str) -> Optional[Product]:
    """
    Get a product by SKU.
    
    Args:
        db: Database session
        sku: Product SKU
        
    Returns:
        Optional[Product]: Product if found, None otherwise
    """
    return db.query(Product).filter(Product.sku == sku).first()


def get_all_products(
    db: Session, 
    skip: int = 0, 
    limit: int = 100,
    category_id: Optional[str] = None,
    active_only: bool = False
) -> List[Product]:
    """
    Get all products with optional filtering.
    
    Args:
        db: Database session
        skip: Number of products to skip
        limit: Maximum number of products to return
        category_id: Optional category ID filter
        active_only: Only include active products
        
    Returns:
        List[Product]: List of products
    """
    query = db.query(Product)
    
    if category_id:
        query = query.filter(Product.category_id == category_id)
    
    if active_only:
        query = query.filter(Product.is_active.is_(True))
    
    return query.offset(skip).limit(limit).all()


def create_product(db: Session, *, obj_in: ProductCreate) -> Product:
    """
    Create a new product.
    
    Args:
        db: Database session
        obj_in: Product creation data
        
    Returns:
        Product: Created product
    """
    db_obj = Product(
        name=obj_in.name,
        description=obj_in.description,
        price=obj_in.price,
        sku=obj_in.sku,
        is_active=obj_in.is_active,
        category_id=obj_in.category_id,
    )
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj


def update_product(
    db: Session, 
    *, 
    db_obj: Product, 
    obj_in: Union[ProductUpdate, Dict[str, Any]]
) -> Product:
    """
    Update a product.
    
    Args:
        db: Database session
        db_obj: Product to update
        obj_in: Product update data
        
    Returns:
        Product: Updated product
    """
    if isinstance(obj_in, dict):
        update_data = obj_in
    else:
        update_data = obj_in.model_dump(exclude_unset=True)
        
    for field in update_data:
        if hasattr(db_obj, field):
            setattr(db_obj, field, update_data[field])
            
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj


def delete_product(db: Session, *, product_id: str) -> Product:
    """
    Delete a product.
    
    Args:
        db: Database session
        product_id: Product ID
        
    Returns:
        Product: Deleted product
    """
    product = db.query(Product).filter(Product.id == product_id).first()
    db.delete(product)
    db.commit()
    return product


# Category CRUD operations
def get_category_by_id(db: Session, category_id: str) -> Optional[Category]:
    """
    Get a category by ID.
    
    Args:
        db: Database session
        category_id: Category ID
        
    Returns:
        Optional[Category]: Category if found, None otherwise
    """
    return db.query(Category).filter(Category.id == category_id).first()


def get_category_by_name(db: Session, name: str) -> Optional[Category]:
    """
    Get a category by name.
    
    Args:
        db: Database session
        name: Category name
        
    Returns:
        Optional[Category]: Category if found, None otherwise
    """
    return db.query(Category).filter(Category.name == name).first()


def get_all_categories(db: Session, skip: int = 0, limit: int = 100) -> List[Category]:
    """
    Get all categories.
    
    Args:
        db: Database session
        skip: Number of categories to skip
        limit: Maximum number of categories to return
        
    Returns:
        List[Category]: List of categories
    """
    return db.query(Category).offset(skip).limit(limit).all()


def create_category(db: Session, *, obj_in: CategoryCreate) -> Category:
    """
    Create a new category.
    
    Args:
        db: Database session
        obj_in: Category creation data
        
    Returns:
        Category: Created category
    """
    db_obj = Category(
        name=obj_in.name,
        description=obj_in.description,
    )
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj


def update_category(
    db: Session, 
    *, 
    db_obj: Category, 
    obj_in: Union[CategoryUpdate, Dict[str, Any]]
) -> Category:
    """
    Update a category.
    
    Args:
        db: Database session
        db_obj: Category to update
        obj_in: Category update data
        
    Returns:
        Category: Updated category
    """
    if isinstance(obj_in, dict):
        update_data = obj_in
    else:
        update_data = obj_in.model_dump(exclude_unset=True)
        
    for field in update_data:
        if hasattr(db_obj, field):
            setattr(db_obj, field, update_data[field])
            
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj


def delete_category(db: Session, *, category_id: str) -> Category:
    """
    Delete a category.
    
    Args:
        db: Database session
        category_id: Category ID
        
    Returns:
        Category: Deleted category
    """
    category = db.query(Category).filter(Category.id == category_id).first()
    db.delete(category)
    db.commit()
    return category