patrickandClaude Sonnet 5
f7367f6df3
Lagerort-Hierarchie (organisation/standort/gebaeude/raum/lager, self-referenzierend), Lagerplätze mit eindeutigem Code, Bestand je Lagerplatz+Material, Ein-/Auslagerung mit lückenlosem Materialbewegungsprotokoll (append-only, analog Lagerbewegung auf Objekt-Ebene, hier aber Material-/Lagerplatz-fein). - Migration 0019: 4 neue Tabellen (lagerort, lagerplatz, bestand, materialbewegung) - Backend: CRUD Lagerort/Lagerplatz, Bestandsabfrage, Ein-/Auslagern-Endpunkte, Negativbestand vorab abgefangen (kein DB-Constraint-Exception-Pfad, gleiches Muster wie objekte.py-Fix vom 2026-09-05) - Frontend: neuer Admin-Tab "Lager" (Lagerort/-platz-Verwaltung, Bestand+Buchung je Lagerplatz) - 2 neue Tests, 139 Tests weiterhin grün WH-005 (Umlagerung) und WH-006 (Inventur) bewusst zurückgestellt - Kern zuerst, laut Nutzer-Entscheidung. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.lager import Bestand, Materialbewegung
|
|
|
|
|
|
class NichtGenugBestandError(Exception):
|
|
pass
|
|
|
|
|
|
async def _bestand_zeile(db: AsyncSession, lagerplatz_id: int, material_id: int) -> Bestand | None:
|
|
result = await db.execute(
|
|
select(Bestand).where(Bestand.lagerplatz_id == lagerplatz_id, Bestand.material_id == material_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def einlagern(
|
|
db: AsyncSession, *, lagerplatz_id: int, material_id: int, menge: Decimal, benutzer_id: int, grund: str | None
|
|
) -> Materialbewegung:
|
|
"""WH-004: Bestand zubuchen (Upsert, falls noch keine Zeile existiert)."""
|
|
bestand = await _bestand_zeile(db, lagerplatz_id, material_id)
|
|
if bestand is None:
|
|
bestand = Bestand(lagerplatz_id=lagerplatz_id, material_id=material_id, menge=Decimal(0))
|
|
db.add(bestand)
|
|
bestand.menge += menge
|
|
|
|
bewegung = Materialbewegung(
|
|
material_id=material_id,
|
|
menge=menge,
|
|
von_lagerplatz_id=None,
|
|
nach_lagerplatz_id=lagerplatz_id,
|
|
benutzer_id=benutzer_id,
|
|
zeitpunkt=datetime.now(timezone.utc),
|
|
grund=grund,
|
|
)
|
|
db.add(bewegung)
|
|
await db.flush()
|
|
return bewegung
|
|
|
|
|
|
async def auslagern(
|
|
db: AsyncSession, *, lagerplatz_id: int, material_id: int, menge: Decimal, benutzer_id: int, grund: str | None
|
|
) -> Materialbewegung:
|
|
"""WH-004: Bestand abbuchen - negativer Bestand ist ausgeschlossen (Akzeptanz-
|
|
kriterium der Kachel), vorab geprüft statt DB-Constraint (gleiches Muster
|
|
wie objekte.py Selbstbezug-Check, Fund 2026-09-05: DB-Exception mitten in
|
|
einer geteilten Session hinterlässt die Session im Zustand DEACTIVE)."""
|
|
bestand = await _bestand_zeile(db, lagerplatz_id, material_id)
|
|
if bestand is None or bestand.menge < menge:
|
|
raise NichtGenugBestandError
|
|
bestand.menge -= menge
|
|
|
|
bewegung = Materialbewegung(
|
|
material_id=material_id,
|
|
menge=menge,
|
|
von_lagerplatz_id=lagerplatz_id,
|
|
nach_lagerplatz_id=None,
|
|
benutzer_id=benutzer_id,
|
|
zeitpunkt=datetime.now(timezone.utc),
|
|
grund=grund,
|
|
)
|
|
db.add(bewegung)
|
|
await db.flush()
|
|
return bewegung
|