from typing import List, Optional from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.product import Product from app.schemas.product import ProductCreate, ProductUpdate class CRUDProduct(CRUDBase[Product, ProductCreate, ProductUpdate]): def search_products( self, db: Session, *, query: str, skip: int = 0, limit: int = 100 ) -> List[Product]: """ Search for products by name or description. Args: db: Database session query: Search query skip: Number of records to skip limit: Maximum number of records to return Returns: List of products matching the search query """ search_pattern = f"%{query}%" return ( db.query(self.model) .filter( (self.model.name.ilike(search_pattern)) | (self.model.description.ilike(search_pattern)) ) .offset(skip) .limit(limit) .all() ) def get_product_by_name(self, db: Session, *, name: str) -> Optional[Product]: """ Get a product by its exact name. Args: db: Database session name: Product name Returns: The product or None if not found """ return db.query(self.model).filter(self.model.name == name).first() def update_stock(self, db: Session, *, product_id: int, quantity: int) -> Product: """ Update product stock quantity. Args: db: Database session product_id: Product ID quantity: Quantity to add (positive) or subtract (negative) Returns: The updated product """ product = self.get(db, id=product_id) if product: product.stock += quantity db.commit() db.refresh(product) return product product = CRUDProduct(Product)