Backend: alle Chargen auflisten + Lagerort mehrerer Chargen auf einmal setzen

GET /lots/rows liefert alle Chargen mit Artikel-Infos (Name, Menge, Einheit,
Lagerort) fuer eine uebergreifende Chargenliste. POST /lots/bulk-location setzt
den Lagerort vieler Chargen in einem Rutsch (reine Ortsangabe, ohne Bewegung).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-29 13:48:19 +02:00
parent 227143d13f
commit 1c88206838
2 changed files with 78 additions and 3 deletions

View File

@@ -1,24 +1,26 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload
from ..crud import product_tracking, resolve_product
from ..database import get_db
from ..deps import get_current_user
from ..models import CategoryTracking, Lot, Movement, MovementType, User
from ..models import CategoryTracking, Location, Lot, Movement, MovementType, User
from ..schemas import (
BatchCheckInRequest,
BatchCheckInResponse,
BulkLotLocation,
CheckInRequest,
CheckInResponse,
CheckOutRequest,
CheckOutResponse,
LotOut,
LotRow,
LotUpdate,
RelocateRequest,
RemoveRequest,
StockActionResponse,
)
from ..services.conversion import ConversionError
from ..services.conversion import ConversionError, display_unit_info
from ..services.dates import clean_precision, normalize_best_before
from ..services.stock import (
StockError,
@@ -236,6 +238,55 @@ def list_lots(
return query.order_by(Lot.best_before.is_(None), Lot.best_before).all()
@router.get("/lots/rows", response_model=list[LotRow])
def list_all_lots(
db: Session = Depends(get_db),
_: User = Depends(get_current_user),
) -> list[LotRow]:
"""Alle Chargen über alle Artikel für die übergreifende Chargenliste mit
Artikelname, Menge und Lagerort. Der Artikel wird eager geladen (kein N+1)."""
lots = (
db.query(Lot)
.options(joinedload(Lot.product))
.order_by(Lot.created_at.desc())
.all()
)
rows: list[LotRow] = []
for lot in lots:
p = lot.product
name, factor = display_unit_info(p)
rows.append(LotRow(
id=lot.id, product_id=p.id, product_name=p.name, product_brand=p.brand,
quantity=lot.quantity, base_unit=p.base_unit,
package_size=p.package_size, package_label=p.package_label,
unit_name=name, unit_factor=factor,
best_before=lot.best_before, best_before_precision=lot.best_before_precision,
location_id=lot.location_id, created_at=lot.created_at,
))
return rows
@router.post("/lots/bulk-location", response_model=int)
def bulk_lot_location(
payload: BulkLotLocation,
db: Session = Depends(get_db),
_: User = Depends(get_current_user),
) -> int:
"""Lagerort mehrerer Chargen auf einmal setzen (oder mit null leeren). Reine
Ortsangabe kein Umbuchen mit Bewegung. Gibt die Anzahl geänderter Zeilen."""
if not payload.lot_ids:
return 0
if payload.location_id is not None and db.get(Location, payload.location_id) is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
n = (
db.query(Lot)
.filter(Lot.id.in_(payload.lot_ids))
.update({Lot.location_id: payload.location_id}, synchronize_session=False)
)
db.commit()
return n
@router.patch("/lots/{lot_id}", response_model=LotOut)
def update_lot(
lot_id: int,