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)