122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
from typing import List, Optional
|
|
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy import desc
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.inventory_movement import InventoryMovement, MovementType
|
|
from app.models.product import Product
|
|
from app.schemas.inventory import InventoryMovementCreate
|
|
from app.utils.uuid import generate_uuid
|
|
|
|
|
|
def get(db: Session, movement_id: str) -> Optional[InventoryMovement]:
|
|
"""
|
|
Get an inventory movement by ID.
|
|
"""
|
|
return db.query(InventoryMovement).filter(InventoryMovement.id == movement_id).first()
|
|
|
|
|
|
def get_multi(
|
|
db: Session,
|
|
*,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
product_id: Optional[str] = None,
|
|
movement_type: Optional[MovementType] = None,
|
|
) -> List[InventoryMovement]:
|
|
"""
|
|
Get multiple inventory movements with optional filtering.
|
|
"""
|
|
query = db.query(InventoryMovement)
|
|
|
|
if product_id:
|
|
query = query.filter(InventoryMovement.product_id == product_id)
|
|
|
|
if movement_type:
|
|
query = query.filter(InventoryMovement.type == movement_type)
|
|
|
|
# Order by most recent first
|
|
query = query.order_by(desc(InventoryMovement.created_at))
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
|
|
def create(
|
|
db: Session, *, obj_in: InventoryMovementCreate, created_by: Optional[str] = None
|
|
) -> InventoryMovement:
|
|
"""
|
|
Create a new inventory movement and update product stock.
|
|
"""
|
|
obj_in_data = jsonable_encoder(obj_in)
|
|
db_obj = InventoryMovement(**obj_in_data, id=generate_uuid(), created_by=created_by)
|
|
|
|
# Start transaction
|
|
try:
|
|
# Add movement record
|
|
db.add(db_obj)
|
|
|
|
# Update product stock based on movement type
|
|
product = (
|
|
db.query(Product).filter(Product.id == obj_in.product_id).with_for_update().first()
|
|
)
|
|
if not product:
|
|
raise ValueError(f"Product with ID {obj_in.product_id} not found")
|
|
|
|
if obj_in.type in [MovementType.STOCK_IN, MovementType.RETURN]:
|
|
product.current_stock += obj_in.quantity
|
|
elif obj_in.type == MovementType.STOCK_OUT:
|
|
if product.current_stock < obj_in.quantity:
|
|
raise ValueError(
|
|
f"Insufficient stock for product {product.name}. Available: {product.current_stock}, Requested: {obj_in.quantity}"
|
|
)
|
|
product.current_stock -= obj_in.quantity
|
|
elif obj_in.type == MovementType.ADJUSTMENT:
|
|
# For adjustments, the quantity value is the delta (can be positive or negative)
|
|
product.current_stock += obj_in.quantity
|
|
|
|
db.add(product)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise e
|
|
|
|
|
|
def remove(db: Session, *, movement_id: str) -> Optional[InventoryMovement]:
|
|
"""
|
|
Delete an inventory movement and revert the stock change.
|
|
This is generally not recommended for production use as it alters inventory history.
|
|
"""
|
|
movement = db.query(InventoryMovement).filter(InventoryMovement.id == movement_id).first()
|
|
if not movement:
|
|
return None
|
|
|
|
# Start transaction
|
|
try:
|
|
# Get the product and prepare to revert the stock change
|
|
product = (
|
|
db.query(Product).filter(Product.id == movement.product_id).with_for_update().first()
|
|
)
|
|
if not product:
|
|
raise ValueError(f"Product with ID {movement.product_id} not found")
|
|
|
|
# Revert the stock change based on movement type
|
|
if movement.type in [MovementType.STOCK_IN, MovementType.RETURN]:
|
|
product.current_stock -= movement.quantity
|
|
elif movement.type == MovementType.STOCK_OUT:
|
|
product.current_stock += movement.quantity
|
|
elif movement.type == MovementType.ADJUSTMENT:
|
|
# For adjustments, reverse the quantity effect
|
|
product.current_stock -= movement.quantity
|
|
|
|
# Delete the movement
|
|
db.delete(movement)
|
|
db.add(product)
|
|
db.commit()
|
|
return movement
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise e
|