233 lines
7.5 KiB
Python
233 lines
7.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from typing import List
|
|
from sqlalchemy.orm import Session
|
|
from .. import models, schemas, auth
|
|
from ..database import get_db
|
|
import json
|
|
|
|
router = APIRouter(
|
|
prefix="/operations",
|
|
tags=["Operations"]
|
|
)
|
|
|
|
@router.post("/check-in", response_model=schemas.Item)
|
|
def check_in_item(
|
|
op: schemas.OperationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Check-in item — only for authenticated users. [M-02] user_id from token."""
|
|
if op.quantity <= 0:
|
|
raise HTTPException(status_code=400, detail="Quantity must be greater than zero")
|
|
|
|
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="Item not found. Register item first.")
|
|
|
|
# Update quantity
|
|
item.quantity += op.quantity
|
|
|
|
# Create Mandatory Audit Log — [M-02] user_id from token
|
|
item_snapshot = {
|
|
"barcode": item.barcode,
|
|
"name": item.name,
|
|
"category": item.category,
|
|
"part_number": item.part_number
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="CHECK_IN",
|
|
target_item_id=item.id,
|
|
target_item_name=item.name,
|
|
target_item_pn=item.part_number,
|
|
target_item_barcode=item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
quantity_change=op.quantity
|
|
)
|
|
|
|
db.add(audit)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
@router.post("/check-out", response_model=schemas.Item)
|
|
def check_out_item(
|
|
op: schemas.OperationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Check-out item — only for authenticated users."""
|
|
if op.quantity <= 0:
|
|
raise HTTPException(status_code=400, detail="Quantity must be greater than zero")
|
|
|
|
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
if item.quantity < op.quantity:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Insufficient stock")
|
|
|
|
# Update quantity
|
|
item.quantity -= op.quantity
|
|
|
|
# Create Mandatory Audit Log
|
|
item_snapshot = {
|
|
"barcode": item.barcode,
|
|
"name": item.name,
|
|
"category": item.category,
|
|
"part_number": item.part_number
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="CHECK_OUT",
|
|
target_item_id=item.id,
|
|
target_item_name=item.name,
|
|
target_item_pn=item.part_number,
|
|
target_item_barcode=item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
quantity_change=-op.quantity
|
|
)
|
|
|
|
db.add(audit)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
@router.post("/trash", response_model=schemas.Item)
|
|
def trash_item(
|
|
op: schemas.TrashOperationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Trash item — only for authenticated users."""
|
|
if op.quantity <= 0:
|
|
raise HTTPException(status_code=400, detail="Quantity must be greater than zero")
|
|
|
|
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
if item.quantity < op.quantity:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Insufficient stock to trash")
|
|
|
|
# Update quantity
|
|
item.quantity -= op.quantity
|
|
|
|
# Create Mandatory Audit Log with TRASH action and reason in details
|
|
item_snapshot = {
|
|
"barcode": item.barcode,
|
|
"name": item.name,
|
|
"category": item.category,
|
|
"part_number": item.part_number
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="TRASH",
|
|
target_item_id=item.id,
|
|
target_item_name=item.name,
|
|
target_item_pn=item.part_number,
|
|
target_item_barcode=item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
quantity_change=-op.quantity,
|
|
details=op.reason
|
|
)
|
|
|
|
db.add(audit)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
@router.post("/bulk-check-out")
|
|
def bulk_check_out(
|
|
bulk_op: schemas.BulkOperationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Bulk check-out — only for authenticated users."""
|
|
results = {"success": [], "errors": []}
|
|
|
|
for op in bulk_op.items:
|
|
try:
|
|
item = db.query(models.Item).filter(models.Item.barcode == op.barcode).first()
|
|
if not item:
|
|
results["errors"].append({"barcode": op.barcode, "error": "Not found"})
|
|
continue
|
|
|
|
if item.quantity < op.quantity:
|
|
results["errors"].append({"barcode": op.barcode, "error": f"Insufficient stock (Available: {item.quantity})"})
|
|
continue
|
|
|
|
# Update quantity
|
|
item.quantity -= op.quantity
|
|
|
|
# Log individual audit for this item
|
|
item_snapshot = {
|
|
"barcode": item.barcode,
|
|
"name": item.name,
|
|
"category": item.category,
|
|
"part_number": item.part_number
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="BULK_CHECK_OUT",
|
|
target_item_id=item.id,
|
|
target_item_name=item.name,
|
|
target_item_pn=item.part_number,
|
|
target_item_barcode=item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
quantity_change=-op.quantity
|
|
)
|
|
db.add(audit)
|
|
results["success"].append({"barcode": op.barcode, "new_quantity": item.quantity})
|
|
|
|
except Exception as e:
|
|
results["errors"].append({"barcode": op.barcode, "error": str(e)})
|
|
|
|
db.commit()
|
|
return results
|
|
|
|
@router.get("/logs", response_model=List[schemas.AuditLogResponse])
|
|
def get_logs(
|
|
limit: int = 50,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Audit logs list — only for authenticated users."""
|
|
# Join with User to get the username directly
|
|
logs_with_users = db.query(
|
|
models.AuditLog.id,
|
|
models.AuditLog.timestamp,
|
|
models.AuditLog.user_id,
|
|
models.User.username,
|
|
models.AuditLog.action,
|
|
models.AuditLog.target_item_id,
|
|
models.AuditLog.target_item_name,
|
|
models.AuditLog.target_item_pn,
|
|
models.AuditLog.target_item_barcode,
|
|
models.AuditLog.target_snapshot,
|
|
models.AuditLog.quantity_change,
|
|
models.AuditLog.details
|
|
).join(models.User, models.AuditLog.user_id == models.User.id).order_by(models.AuditLog.timestamp.desc()).limit(limit).all()
|
|
|
|
# Mapper to dictionary for Pydantic
|
|
return [
|
|
{
|
|
"id": l.id,
|
|
"timestamp": l.timestamp,
|
|
"user_id": l.user_id,
|
|
"username": l.username,
|
|
"action": l.action,
|
|
"target_item_id": l.target_item_id,
|
|
"target_item_name": l.target_item_name,
|
|
"target_item_pn": l.target_item_pn,
|
|
"target_item_barcode": l.target_item_barcode,
|
|
"target_snapshot": l.target_snapshot,
|
|
"quantity_change": l.quantity_change,
|
|
"details": l.details
|
|
} for l in logs_with_users
|
|
]
|