fix(backend): Hersteller/Modell-Duplikatprüfung vorab statt über UNIQUE-Constraint
CI / backend-tests (push) Successful in 2m17s
CI / frontend-build (push) Successful in 27s

Testsuite-get_db-Override macht kein Rollback bei Exceptions (dokumentierter
Fund 2026-09-05) - ein Flush-Fehlschlag über IntegrityError hinterlässt die
geteilte Test-Session im PendingRollbackError-Zustand für jeden folgenden
Request im selben Test. Projekt-Konvention laut geraet_instanz.py::erstelle
ist deshalb explizites Vorab-Prüfen statt Verlassen auf den DB-Constraint -
hier nachgezogen.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-08 14:56:05 +02:00
co-authored by Claude Sonnet 5
parent 2dd7e574e6
commit 08ce8efe55
+15 -9
View File
@@ -1,6 +1,5 @@
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import func, select from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user, require_roles, require_roles_or_permission from app.api.deps import get_current_user, require_roles, require_roles_or_permission
@@ -96,12 +95,17 @@ async def erstelle_hersteller(
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
_=Depends(require_roles(RolleTyp.administration)), _=Depends(require_roles(RolleTyp.administration)),
) -> Hersteller: ) -> Hersteller:
# Explizit vorab prüfen statt sich auf den UNIQUE-Constraint zu verlassen:
# ein Flush-Fehlschlag hinterlässt die Session im Zustand DEACTIVE
# (SQLAlchemy 2.0), was in einer über mehrere Requests geteilten Session
# (z.B. Testsuite) jeden folgenden Request mit PendingRollbackError zum
# Absturz bringt (gleicher Fund wie bei geraet_instanz.py::erstelle).
bestehender = await db.execute(select(Hersteller).where(Hersteller.name == payload.name))
if bestehender.scalar_one_or_none() is not None:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Hersteller existiert bereits")
hersteller = Hersteller(**payload.model_dump()) hersteller = Hersteller(**payload.model_dump())
db.add(hersteller) db.add(hersteller)
try: await db.flush()
await db.flush()
except IntegrityError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Hersteller existiert bereits") from exc
return hersteller return hersteller
@@ -126,12 +130,14 @@ async def erstelle_modell(
) -> Modell: ) -> Modell:
if await db.get(Hersteller, payload.hersteller_id) is None: if await db.get(Hersteller, payload.hersteller_id) is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden")
bestehendes = await db.execute(
select(Modell).where(Modell.hersteller_id == payload.hersteller_id, Modell.name == payload.name)
)
if bestehendes.scalar_one_or_none() is not None:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Modell existiert bereits")
modell = Modell(**payload.model_dump()) modell = Modell(**payload.model_dump())
db.add(modell) db.add(modell)
try: await db.flush()
await db.flush()
except IntegrityError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Modell existiert bereits") from exc
return modell return modell