from typing import List from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.stock_transaction import StockTransaction, TransactionType from app.schemas.stock_transaction import StockTransactionCreate from app.crud.item import item as item_crud class CRUDStockTransaction(CRUDBase[StockTransaction, StockTransactionCreate, None]): def create_with_stock_update( self, db: Session, *, obj_in: StockTransactionCreate ) -> StockTransaction: # Create the transaction transaction = self.create(db, obj_in=obj_in) # Update item stock db_item = item_crud.get(db, id=obj_in.item_id) if db_item: if obj_in.transaction_type == TransactionType.IN: db_item.quantity_in_stock += obj_in.quantity elif obj_in.transaction_type == TransactionType.OUT: db_item.quantity_in_stock -= obj_in.quantity elif obj_in.transaction_type == TransactionType.ADJUSTMENT: db_item.quantity_in_stock = obj_in.quantity db.add(db_item) db.commit() return transaction def get_by_item( self, db: Session, *, item_id: int, skip: int = 0, limit: int = 100 ) -> List[StockTransaction]: return ( db.query(StockTransaction) .filter(StockTransaction.item_id == item_id) .order_by(StockTransaction.created_at.desc()) .offset(skip) .limit(limit) .all() ) def get_by_type( self, db: Session, *, transaction_type: TransactionType, skip: int = 0, limit: int = 100, ) -> List[StockTransaction]: return ( db.query(StockTransaction) .filter(StockTransaction.transaction_type == transaction_type) .order_by(StockTransaction.created_at.desc()) .offset(skip) .limit(limit) .all() ) stock_transaction = CRUDStockTransaction(StockTransaction)