patrickandClaude Sonnet 5
06baf03708
Verwendung je Material einsehbar (Objektposition/Vorlagenposition/
Bestand/Kontrollposition/Fehlbestand/Ausgabe/Nachfüllung/
Materialbewegung), POST /materialien/{id}/ersetzen hängt alle
Referenzen auf ein Ziel-Material um und löscht die Quelle. Bricht bei
Konflikten ab (z.B. dasselbe Objekt hätte danach zwei Positionen für
dasselbe Material) statt Daten stillschweigend zu verlieren. Admin-UI:
"Zusammenführen" je Material zeigt Verwendung + Ziel-Auswahl.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
64 lines
3.0 KiB
Python
64 lines
3.0 KiB
Python
from sqlalchemy import func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.ausgabe import Ausgabe
|
|
from app.models.fehlbestand import Fehlbestand
|
|
from app.models.kontrolle import Kontrollposition
|
|
from app.models.lager import Bestand, Materialbewegung
|
|
from app.models.nachfuellung import Nachfuellung
|
|
from app.models.objektposition import Objektposition
|
|
from app.models.vorlage import Vorlagenposition
|
|
|
|
# (Modell, Gruppierungs-Spalte für den Konflikt-Check, Anzeige-Name)
|
|
_KONFLIKT_TABELLEN = [
|
|
(Objektposition, Objektposition.objekt_id, "Objektposition"),
|
|
(Vorlagenposition, Vorlagenposition.vorlage_id, "Vorlagenposition"),
|
|
(Bestand, Bestand.lagerplatz_id, "Bestand"),
|
|
(Kontrollposition, Kontrollposition.kontrolle_id, "Kontrollposition"),
|
|
]
|
|
# Reine Protokoll-/Bewegungstabellen ohne Eindeutigkeits-Constraint auf
|
|
# material_id - können immer gefahrlos umgehängt werden.
|
|
_SICHERE_TABELLEN = [Fehlbestand, Ausgabe, Nachfuellung, Materialbewegung]
|
|
|
|
|
|
async def hole_verwendung(db: AsyncSession, material_id: int) -> dict[str, int]:
|
|
verwendung = {}
|
|
for model, _, name in _KONFLIKT_TABELLEN:
|
|
result = await db.execute(select(func.count()).select_from(model).where(model.material_id == material_id))
|
|
verwendung[name] = result.scalar_one()
|
|
for model in _SICHERE_TABELLEN:
|
|
result = await db.execute(select(func.count()).select_from(model).where(model.material_id == material_id))
|
|
verwendung[model.__tablename__] = result.scalar_one()
|
|
return verwendung
|
|
|
|
|
|
async def finde_konflikte(db: AsyncSession, *, quelle_id: int, ziel_id: int) -> list[str]:
|
|
"""Für Tabellen mit UNIQUE(gruppen_spalte, material_id) darf nach dem
|
|
Umhängen nicht dieselbe Kombination doppelt entstehen - z. B. wenn ein
|
|
Objekt schon eine Position mit dem Ziel-Material hat UND eine mit dem
|
|
Quell-Material."""
|
|
konflikte: list[str] = []
|
|
for model, gruppen_spalte, name in _KONFLIKT_TABELLEN:
|
|
quelle_gruppen = (
|
|
(await db.execute(select(gruppen_spalte).where(model.material_id == quelle_id))).scalars().all()
|
|
)
|
|
if not quelle_gruppen:
|
|
continue
|
|
ziel_gruppen = set(
|
|
(await db.execute(select(gruppen_spalte).where(model.material_id == ziel_id))).scalars().all()
|
|
)
|
|
ueberschneidung = set(quelle_gruppen) & ziel_gruppen
|
|
if ueberschneidung:
|
|
konflikte.append(f"{name}: {len(ueberschneidung)} Konflikt(e) - beide Materialien dort gleichzeitig vorhanden")
|
|
return konflikte
|
|
|
|
|
|
async def fuehre_zusammen(db: AsyncSession, *, quelle_id: int, ziel_id: int) -> None:
|
|
"""Voraussetzung: finde_konflikte() wurde vorher aufgerufen und war leer -
|
|
der Aufrufer (Endpunkt) verweigert den Merge sonst."""
|
|
for model, _, _ in _KONFLIKT_TABELLEN:
|
|
await db.execute(update(model).where(model.material_id == quelle_id).values(material_id=ziel_id))
|
|
for model in _SICHERE_TABELLEN:
|
|
await db.execute(update(model).where(model.material_id == quelle_id).values(material_id=ziel_id))
|
|
await db.flush()
|