44 lines
1.1 KiB
Python
44 lines
1.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from app.models.product import Product
|
|
from app.schemas.product import ProductCreate, ProductUpdate
|
|
|
|
|
|
def create_product(db: Session, data: ProductCreate) -> Product:
|
|
prod = Product(**data.model_dump())
|
|
db.add(prod)
|
|
db.commit()
|
|
db.refresh(prod)
|
|
return prod
|
|
|
|
|
|
def get_products(db: Session, skip: int = 0, limit: int = 20):
|
|
return db.query(Product).offset(skip).limit(limit).all()
|
|
|
|
|
|
def get_product(db: Session, product_id: str):
|
|
return db.query(Product).filter(Product.id == product_id).first()
|
|
|
|
|
|
def update_product(db: Session, product_id: str, data: ProductUpdate):
|
|
prod = get_product(db, product_id)
|
|
if not prod:
|
|
return None
|
|
for field, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(prod, field, value)
|
|
db.commit()
|
|
db.refresh(prod)
|
|
return prod
|
|
|
|
|
|
def delete_product(db: Session, product_id: str):
|
|
prod = get_product(db, product_id)
|
|
if not prod:
|
|
return False
|
|
db.delete(prod)
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def get_low_stock_products(db: Session):
|
|
return db.query(Product).filter(Product.current_stock < Product.min_stock).all()
|