89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from app.db.session import get_db
|
|
from app.models.product import Product
|
|
from app.models.stock_movement import StockMovement, MovementType
|
|
from app.schemas.stock_movement import StockMovement as StockMovementSchema, StockMovementCreate
|
|
from app.schemas.product import Product as ProductSchema
|
|
|
|
router = APIRouter()
|
|
|
|
@router.post("/stock-movement", response_model=StockMovementSchema)
|
|
def create_stock_movement(movement: StockMovementCreate, db: Session = Depends(get_db)):
|
|
product = db.query(Product).filter(Product.id == movement.product_id).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="Product not found")
|
|
|
|
db_movement = StockMovement(**movement.dict())
|
|
db.add(db_movement)
|
|
|
|
if movement.movement_type == MovementType.IN:
|
|
product.quantity_in_stock += movement.quantity
|
|
elif movement.movement_type == MovementType.OUT:
|
|
if product.quantity_in_stock < movement.quantity:
|
|
raise HTTPException(status_code=400, detail="Insufficient stock")
|
|
product.quantity_in_stock -= movement.quantity
|
|
elif movement.movement_type == MovementType.ADJUSTMENT:
|
|
product.quantity_in_stock = movement.quantity
|
|
|
|
db.commit()
|
|
db.refresh(db_movement)
|
|
return db_movement
|
|
|
|
@router.get("/stock-movements", response_model=List[StockMovementSchema])
|
|
def read_stock_movements(skip: int = 0, limit: int = 100, product_id: int = None, db: Session = Depends(get_db)):
|
|
query = db.query(StockMovement)
|
|
if product_id:
|
|
query = query.filter(StockMovement.product_id == product_id)
|
|
movements = query.order_by(StockMovement.created_at.desc()).offset(skip).limit(limit).all()
|
|
return movements
|
|
|
|
@router.get("/low-stock", response_model=List[ProductSchema])
|
|
def get_low_stock_products(db: Session = Depends(get_db)):
|
|
products = db.query(Product).filter(
|
|
Product.quantity_in_stock <= Product.min_stock_level,
|
|
Product.is_active == True
|
|
).all()
|
|
return products
|
|
|
|
@router.get("/stock-report")
|
|
def get_stock_report(db: Session = Depends(get_db)):
|
|
total_products = db.query(Product).filter(Product.is_active == True).count()
|
|
low_stock_count = db.query(Product).filter(
|
|
Product.quantity_in_stock <= Product.min_stock_level,
|
|
Product.is_active == True
|
|
).count()
|
|
out_of_stock_count = db.query(Product).filter(
|
|
Product.quantity_in_stock == 0,
|
|
Product.is_active == True
|
|
).count()
|
|
|
|
total_stock_value = db.query(Product).filter(Product.is_active == True).all()
|
|
total_value = sum(p.quantity_in_stock * (p.cost or 0) for p in total_stock_value)
|
|
|
|
return {
|
|
"total_products": total_products,
|
|
"low_stock_products": low_stock_count,
|
|
"out_of_stock_products": out_of_stock_count,
|
|
"total_stock_value": total_value
|
|
}
|
|
|
|
@router.put("/adjust-stock/{product_id}")
|
|
def adjust_stock(product_id: int, new_quantity: int, notes: str = None, db: Session = Depends(get_db)):
|
|
product = db.query(Product).filter(Product.id == product_id).first()
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="Product not found")
|
|
|
|
movement = StockMovement(
|
|
product_id=product_id,
|
|
movement_type=MovementType.ADJUSTMENT,
|
|
quantity=new_quantity,
|
|
notes=notes or f"Stock adjusted to {new_quantity}"
|
|
)
|
|
|
|
db.add(movement)
|
|
product.quantity_in_stock = new_quantity
|
|
db.commit()
|
|
|
|
return {"message": f"Stock adjusted to {new_quantity}", "new_quantity": new_quantity} |