patrickandClaude Sonnet 5
6934276e8a
Backend: geraet_instanz-CRUD (Liste/Anlegen/Status ändern inkl. optionalem Fehlbestand pro Gerät/Löschen), pruefintervall_monate in Objektposition- Schemas verdrahtet (Prüfdatum -> naechste_pruefung wird automatisch berechnet, Monatsrechnung ohne neue Dependency). Frontend: GeraeteInstanzenListe ersetzt das Einzel-SN-Feld im Admin-Portal (ObjektPositionenPanel) durch eine Instanz-Liste mit Status/Prüfdatum/ Bemerkung je Exemplar - löst das Karte-14-Kernproblem (mehrere Geräte pro Position). ObjektSection: Anhänger-Kopplung bei Anlage und nachträglich änderbar. Kontroll-Erfassung (PositionCard/KontrollPage) bleibt unverändert - die komplett neue Erfassungs-UI für geraet_sn (Umsetzungsschritt 4, Karte 14) ist bewusst nicht Teil dieses Commits. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import require_roles
|
|
from app.db.session import get_db
|
|
from app.models.auth import KnotenTyp, RolleTyp, Systemknoten
|
|
from app.models.geraet_instanz import GeraetInstanz
|
|
from app.models.objektposition import Objektposition
|
|
from app.schemas.geraet_instanz import GeraetInstanzCreate, GeraetInstanzRead, GeraetInstanzStatusUpdate
|
|
from app.services.geraet_instanz import aendere_status, erstelle, liste_fuer_position, loesche
|
|
|
|
router = APIRouter()
|
|
|
|
_materialverantwortliche = require_roles(
|
|
RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
|
|
)
|
|
|
|
|
|
async def _hauptserver_id(db: AsyncSession) -> int:
|
|
result = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
|
|
return result.scalar_one()
|
|
|
|
|
|
@router.get("/objektpositionen/{position_id}/geraete", response_model=list[GeraetInstanzRead])
|
|
async def liste_geraete(
|
|
position_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
|
|
) -> list[GeraetInstanz]:
|
|
return await liste_fuer_position(db, position_id)
|
|
|
|
|
|
@router.post(
|
|
"/objektpositionen/{position_id}/geraete",
|
|
response_model=GeraetInstanzRead,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def erstelle_geraet(
|
|
position_id: uuid.UUID,
|
|
payload: GeraetInstanzCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_=Depends(_materialverantwortliche),
|
|
) -> GeraetInstanz:
|
|
"""Karte 14: löst objektposition.seriennummer als Einzelfeld ab - beliebig
|
|
viele Geräte-Exemplare pro Position."""
|
|
position = await db.get(Objektposition, position_id)
|
|
if position is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objektposition nicht gefunden")
|
|
try:
|
|
return await erstelle(db, objektposition_id=position_id, seriennummer=payload.seriennummer)
|
|
except IntegrityError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT, detail="Seriennummer an dieser Position bereits vergeben"
|
|
) from exc
|
|
|
|
|
|
@router.patch("/geraete/{geraet_id}", response_model=GeraetInstanzRead)
|
|
async def aendere_geraet_status(
|
|
geraet_id: uuid.UUID,
|
|
payload: GeraetInstanzStatusUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user=Depends(_materialverantwortliche),
|
|
) -> GeraetInstanz:
|
|
instanz = await db.get(GeraetInstanz, geraet_id)
|
|
if instanz is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
|
|
zustaendiger_server_id = await _hauptserver_id(db)
|
|
aktualisiert, _fehlbestand_id = await aendere_status(
|
|
db,
|
|
instanz=instanz,
|
|
status=payload.status,
|
|
pruefdatum=payload.pruefdatum,
|
|
bemerkung=payload.bemerkung,
|
|
erzeuge_fehlbestand=payload.erzeuge_fehlbestand,
|
|
benutzer_id=current_user.id,
|
|
zustaendiger_server_id=zustaendiger_server_id,
|
|
)
|
|
return aktualisiert
|
|
|
|
|
|
@router.delete("/geraete/{geraet_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def loesche_geraet(
|
|
geraet_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
|
|
) -> None:
|
|
instanz = await db.get(GeraetInstanz, geraet_id)
|
|
if instanz is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
await loesche(db, instanz)
|