from typing import List, Optional from sqlalchemy.orm import Session from sqlalchemy import desc from app.models.product import Product from app.schemas.product import ProductCreate, ProductUpdate def get_product(db: Session, product_id: int) -> Optional[Product]: return db.query(Product).filter(Product.id == product_id).first() def get_product_by_sku(db: Session, sku: str) -> Optional[Product]: return db.query(Product).filter(Product.sku == sku).first() def get_products( db: Session, skip: int = 0, limit: int = 100, category: Optional[str] = None ) -> List[Product]: query = db.query(Product) if category: query = query.filter(Product.category == category) return query.order_by(desc(Product.created_at)).offset(skip).limit(limit).all() def create_product(db: Session, product: ProductCreate) -> Product: db_product = Product( name=product.name, description=product.description, price=product.price, sku=product.sku, stock=product.stock, category=product.category ) db.add(db_product) db.commit() db.refresh(db_product) return db_product def update_product( db: Session, db_product: Product, product_update: ProductUpdate ) -> Product: update_data = product_update.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(db_product, field, value) db.commit() db.refresh(db_product) return db_product def delete_product(db: Session, db_product: Product) -> None: db.delete(db_product) db.commit()