b49546b297
Bestehende Lagerorte lassen sich jetzt nachträglich einem anderen Elternort zuordnen oder auf die oberste Ebene holen. - Backend: LocationUpdate um parent_id (Name jetzt optional); update_location haengt um, mit Schutz gegen Ringe (weder auf sich selbst noch auf einen eigenen Unterort) und Existenzpruefung des Ziels. 6 Tests. - Web: Lagerorte-Seite bekommt je Ort Bearbeiten (Name + Elternort ueber den CategorySelect-Baum, eigene Unterorte ausgeschlossen); api.updateLocation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from sqlalchemy import func
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..database import get_db
|
||
from ..deps import get_current_user, require_admin
|
||
from ..models import Location, User
|
||
from ..schemas import LocationCreate, LocationOut, LocationUpdate
|
||
|
||
router = APIRouter(prefix="/locations", tags=["locations"])
|
||
|
||
|
||
def _descendant_ids(db: Session, location_id: int) -> set[int]:
|
||
"""Alle Unterorte (rekursiv) – als Ziel beim Umhängen ausgeschlossen, sonst
|
||
entstünde ein Ring."""
|
||
result: set[int] = set()
|
||
stack = [location_id]
|
||
while stack:
|
||
cur = stack.pop()
|
||
for kid in db.query(Location).filter(Location.parent_id == cur).all():
|
||
if kid.id not in result:
|
||
result.add(kid.id)
|
||
stack.append(kid.id)
|
||
return result
|
||
|
||
|
||
@router.get("", response_model=list[LocationOut])
|
||
def list_locations(
|
||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||
) -> list[Location]:
|
||
return db.query(Location).order_by(Location.name).all()
|
||
|
||
|
||
@router.post("", response_model=LocationOut, status_code=status.HTTP_201_CREATED)
|
||
def create_location(
|
||
payload: LocationCreate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> Location:
|
||
loc = Location(name=payload.name, parent_id=payload.parent_id)
|
||
db.add(loc)
|
||
db.commit()
|
||
db.refresh(loc)
|
||
return loc
|
||
|
||
|
||
@router.patch("/{location_id}", response_model=LocationOut)
|
||
def update_location(
|
||
location_id: int,
|
||
payload: LocationUpdate,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> Location:
|
||
"""Umbenennen und/oder umhängen. Chargen haengen an der ID, behalten ihren
|
||
Lagerort also."""
|
||
loc = db.get(Location, location_id)
|
||
if loc is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
|
||
|
||
data = payload.model_dump(exclude_unset=True)
|
||
|
||
if "name" in data and data["name"]:
|
||
name = data["name"].strip()
|
||
doppelt = (
|
||
db.query(Location)
|
||
.filter(func.lower(Location.name) == name.lower(), Location.id != location_id)
|
||
.first()
|
||
)
|
||
if doppelt is not None:
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "Diesen Lagerort gibt es bereits")
|
||
loc.name = name
|
||
|
||
# parent_id nur anfassen, wenn ausdrücklich mitgeschickt (None = oberste Ebene).
|
||
if "parent_id" in data:
|
||
neu = data["parent_id"]
|
||
if neu is not None:
|
||
if db.get(Location, neu) is None:
|
||
raise HTTPException(
|
||
status.HTTP_404_NOT_FOUND, "Übergeordneter Lagerort nicht gefunden"
|
||
)
|
||
if neu == location_id or neu in _descendant_ids(db, location_id):
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST,
|
||
"Ein Lagerort kann nicht sich selbst oder einem seiner Unterorte "
|
||
"untergeordnet werden.",
|
||
)
|
||
loc.parent_id = neu
|
||
|
||
db.commit()
|
||
db.refresh(loc)
|
||
return loc
|
||
|
||
|
||
@router.delete("/{location_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_location(
|
||
location_id: int,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> None:
|
||
loc = db.get(Location, location_id)
|
||
if loc is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
|
||
db.delete(loc)
|
||
db.commit()
|