37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.product import Product
|
|
from app.schemas.product import ProductCreate, ProductUpdate
|
|
from app.services.base import CRUDBase
|
|
|
|
|
|
class ProductService(CRUDBase[Product, ProductCreate, ProductUpdate]):
|
|
def get_active(self, db: Session, *, id: int) -> Optional[Product]:
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.id == id, Product.is_active)
|
|
.first()
|
|
)
|
|
|
|
def get_multi_active(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[Product]:
|
|
return (
|
|
db.query(Product)
|
|
.filter(Product.is_active)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def update_stock(self, db: Session, *, db_obj: Product, new_stock: int) -> Product:
|
|
db_obj.stock = new_stock
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
product_service = ProductService(Product)
|