feat(dokumente): DOC-002 feste Dokumenttypen statt Freitext
Neues Pflichtfeld dokumenttyp (Enum: Prüfprotokoll/Wartungsbericht/ Bedienungsanleitung/Rechnung/Zulassungsdokument/Sonstiges, Migration 0026) - macht Dokumente kategorisier- und filterbar statt nur per Freitext- Beschreibung auffindbar zu sein. Backend: Pflichtfeld beim Upload, optionaler Query-Filter bei GET /dokumente. Frontend: Auswahl-Dropdown beim Upload, Typ-Badge + Filter-Dropdown in der Liste (DokumentePanel). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from app.api.deps import get_current_user, require_roles
|
||||
from app.db.session import get_db
|
||||
from app.models.auth import RolleTyp
|
||||
from app.models.dokument import Dokument
|
||||
from app.models.dokument import Dokument, DokumentTyp
|
||||
from app.schemas.dokument import DokumentRead, EntitaetTyp
|
||||
from app.services.dokument import (
|
||||
DateityperlaubtError,
|
||||
@@ -36,16 +36,20 @@ _materialverantwortliche = require_roles(
|
||||
async def liste_dokumente(
|
||||
entitaet_typ: EntitaetTyp,
|
||||
entitaet_id: str,
|
||||
dokumenttyp: DokumentTyp | None = None,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
_=Depends(get_current_user),
|
||||
) -> list[Dokument]:
|
||||
return await liste_fuer_entitaet(db, entitaet_typ=entitaet_typ, entitaet_id=entitaet_id)
|
||||
return await liste_fuer_entitaet(
|
||||
db, entitaet_typ=entitaet_typ, entitaet_id=entitaet_id, dokumenttyp=dokumenttyp
|
||||
)
|
||||
|
||||
|
||||
@router.post("/dokumente", response_model=DokumentRead, status_code=status.HTTP_201_CREATED)
|
||||
async def lade_dokument_hoch(
|
||||
entitaet_typ: EntitaetTyp = Form(...),
|
||||
entitaet_id: str = Form(...),
|
||||
dokumenttyp: DokumentTyp = Form(...),
|
||||
beschreibung: str | None = Form(None),
|
||||
datei: UploadFile = File(...),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
@@ -60,6 +64,7 @@ async def lade_dokument_hoch(
|
||||
dateiname=datei.filename or "unbenannt",
|
||||
mime_type=datei.content_type or "application/octet-stream",
|
||||
inhalt=inhalt,
|
||||
dokumenttyp=dokumenttyp,
|
||||
beschreibung=beschreibung,
|
||||
hochgeladen_von=current_user.id,
|
||||
)
|
||||
|
||||
@@ -1,13 +1,30 @@
|
||||
import enum
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import ForeignKey, Integer, String
|
||||
from sqlalchemy.dialects.postgresql import TIMESTAMP, UUID
|
||||
from sqlalchemy.dialects.postgresql import ENUM as PgEnum, TIMESTAMP, UUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class DokumentTyp(str, enum.Enum):
|
||||
"""DOC-002: feste Kategorie statt Freitext - macht Dokumente filter-/
|
||||
wiederfindbar (Nutzer-Vorgabe Modul 10: Prüfprotokolle/Wartungsberichte/
|
||||
Bedienungsanleitungen/Rechnungen/Zulassungsdokumente)."""
|
||||
|
||||
pruefprotokoll = "pruefprotokoll"
|
||||
wartungsbericht = "wartungsbericht"
|
||||
bedienungsanleitung = "bedienungsanleitung"
|
||||
rechnung = "rechnung"
|
||||
zulassungsdokument = "zulassungsdokument"
|
||||
sonstiges = "sonstiges"
|
||||
|
||||
|
||||
dokument_typ_pg = PgEnum(DokumentTyp, name="dokument_typ", create_type=False)
|
||||
|
||||
|
||||
class Dokument(Base):
|
||||
"""Roadmap Phase 5 (Modul Dokumente): polymorpher Datei-Anhang an beliebige
|
||||
Ressource (Objekt, Objektposition, Geräteinstanz, Mangel, Fahrzeugdetails,
|
||||
@@ -30,6 +47,9 @@ class Dokument(Base):
|
||||
# Nullable wegen Altbestand vor dieser Migration (Hash nachträglich ohne
|
||||
# Dateizugriff nicht sinnvoll befüllbar); jeder neue Upload setzt ihn.
|
||||
dateihash: Mapped[str | None] = mapped_column(String(64))
|
||||
dokumenttyp: Mapped[DokumentTyp] = mapped_column(
|
||||
dokument_typ_pg, nullable=False, default=DokumentTyp.sonstiges
|
||||
)
|
||||
beschreibung: Mapped[str | None] = mapped_column(String)
|
||||
hochgeladen_von: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
|
||||
hochgeladen_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
|
||||
|
||||
@@ -4,6 +4,8 @@ from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from app.models.dokument import DokumentTyp
|
||||
|
||||
# Bewusst geschlossene Liste statt Freitext (Konsistenz mit Mangel.entitaet_typ-
|
||||
# Validierung) - jeder Ressourcentyp, an den Dokumente angehängt werden dürfen,
|
||||
# muss hier explizit freigeschaltet werden.
|
||||
@@ -20,6 +22,7 @@ class DokumentRead(BaseModel):
|
||||
dateiname: str
|
||||
mime_type: str
|
||||
groesse_bytes: int
|
||||
dokumenttyp: DokumentTyp
|
||||
beschreibung: str | None
|
||||
hochgeladen_von: int
|
||||
hochgeladen_am: datetime
|
||||
|
||||
@@ -8,7 +8,7 @@ from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.app_settings import settings
|
||||
from app.models.dokument import Dokument
|
||||
from app.models.dokument import Dokument, DokumentTyp
|
||||
|
||||
# Erlaubte MIME-Types (Nutzer-Vorgabe Modul 10: PDF/Bilder/Prüfprotokolle/
|
||||
# Wartungsberichte/Bedienungsanleitungen/Rechnungen/Zulassungsdokumente) -
|
||||
@@ -52,6 +52,7 @@ async def speichere_dokument(
|
||||
dateiname: str,
|
||||
mime_type: str,
|
||||
inhalt: bytes,
|
||||
dokumenttyp: DokumentTyp,
|
||||
beschreibung: str | None,
|
||||
hochgeladen_von: int,
|
||||
) -> Dokument:
|
||||
@@ -87,6 +88,7 @@ async def speichere_dokument(
|
||||
mime_type=mime_type,
|
||||
groesse_bytes=len(inhalt),
|
||||
dateihash=dateihash,
|
||||
dokumenttyp=dokumenttyp,
|
||||
beschreibung=beschreibung,
|
||||
hochgeladen_von=hochgeladen_von,
|
||||
hochgeladen_am=datetime.now(timezone.utc),
|
||||
@@ -96,12 +98,13 @@ async def speichere_dokument(
|
||||
return dokument
|
||||
|
||||
|
||||
async def liste_fuer_entitaet(db: AsyncSession, *, entitaet_typ: str, entitaet_id: str) -> list[Dokument]:
|
||||
result = await db.execute(
|
||||
select(Dokument)
|
||||
.where(Dokument.entitaet_typ == entitaet_typ, Dokument.entitaet_id == entitaet_id)
|
||||
.order_by(Dokument.hochgeladen_am.desc())
|
||||
)
|
||||
async def liste_fuer_entitaet(
|
||||
db: AsyncSession, *, entitaet_typ: str, entitaet_id: str, dokumenttyp: DokumentTyp | None = None
|
||||
) -> list[Dokument]:
|
||||
bedingungen = [Dokument.entitaet_typ == entitaet_typ, Dokument.entitaet_id == entitaet_id]
|
||||
if dokumenttyp is not None:
|
||||
bedingungen.append(Dokument.dokumenttyp == dokumenttyp)
|
||||
result = await db.execute(select(Dokument).where(*bedingungen).order_by(Dokument.hochgeladen_am.desc()))
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user