from typing import List, Optional from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.inventory_item import InventoryItem from app.schemas.inventory_item import InventoryItemCreate, InventoryItemUpdate class CRUDInventoryItem(CRUDBase[InventoryItem, InventoryItemCreate, InventoryItemUpdate]): def get_by_sku(self, db: Session, *, sku: str) -> Optional[InventoryItem]: return db.query(InventoryItem).filter(InventoryItem.sku == sku).first() def get_by_barcode(self, db: Session, *, barcode: str) -> Optional[InventoryItem]: return db.query(InventoryItem).filter(InventoryItem.barcode == barcode).first() def get_low_stock_items(self, db: Session) -> List[InventoryItem]: return db.query(InventoryItem).filter( InventoryItem.quantity_in_stock <= InventoryItem.minimum_stock_level ).all() def search_by_name(self, db: Session, *, name: str, skip: int = 0, limit: int = 100) -> List[InventoryItem]: return db.query(InventoryItem).filter( InventoryItem.name.contains(name) ).offset(skip).limit(limit).all() def get_by_category(self, db: Session, *, category_id: int, skip: int = 0, limit: int = 100) -> List[InventoryItem]: return db.query(InventoryItem).filter( InventoryItem.category_id == category_id ).offset(skip).limit(limit).all() def get_by_supplier(self, db: Session, *, supplier_id: int, skip: int = 0, limit: int = 100) -> List[InventoryItem]: return db.query(InventoryItem).filter( InventoryItem.supplier_id == supplier_id ).offset(skip).limit(limit).all() inventory_item = CRUDInventoryItem(InventoryItem)