feat(akte): FILE-009 Freitext-Notizen an beliebiger Ressource
Polymorphes akte_notiz-Modell (gleiches entitaet_typ/entitaet_id-Muster wie Dokument/Historie/Mangel), CRUD-Endpunkte, NotizenPanel im Akte-Grid. Löschen nur durch Autor oder Administrator. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
@@ -0,0 +1,34 @@
|
||||
"""FILE-009: Freitext-Notizen in der Akte.
|
||||
|
||||
Revision ID: 0035_akte_notiz
|
||||
Revises: 0034_person
|
||||
Create Date: 2026-09-10
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0035_akte_notiz"
|
||||
down_revision: Union[str, None] = "0034_person"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute(
|
||||
"""
|
||||
CREATE TABLE akte_notiz (
|
||||
id UUID PRIMARY KEY,
|
||||
entitaet_typ TEXT NOT NULL,
|
||||
entitaet_id TEXT NOT NULL,
|
||||
text TEXT NOT NULL,
|
||||
autor_id INTEGER NOT NULL REFERENCES benutzer(id),
|
||||
erstellt_am TIMESTAMPTZ NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
op.execute("CREATE INDEX ix_akte_notiz_entitaet ON akte_notiz (entitaet_typ, entitaet_id)")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("DROP TABLE akte_notiz")
|
||||
@@ -16,6 +16,7 @@ from app.api.v1.endpoints import (
|
||||
lager,
|
||||
lagerbewegung,
|
||||
mangel,
|
||||
notiz,
|
||||
objekte,
|
||||
permission,
|
||||
personal,
|
||||
@@ -44,6 +45,7 @@ api_router.include_router(personal.router, tags=["personal"])
|
||||
api_router.include_router(mangel.router, tags=["mangel"])
|
||||
api_router.include_router(lagerbewegung.router, tags=["lagerbewegung"])
|
||||
api_router.include_router(dokument.router, tags=["dokument"])
|
||||
api_router.include_router(notiz.router, tags=["notiz"])
|
||||
api_router.include_router(permission.router, tags=["permission"])
|
||||
api_router.include_router(akte.router, tags=["akte"])
|
||||
api_router.include_router(lager.router, tags=["lager"])
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
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 Benutzer, RolleTyp
|
||||
from app.models.notiz import AkteNotiz
|
||||
from app.schemas.dokument import EntitaetTyp
|
||||
from app.schemas.notiz import AkteNotizCreate, AkteNotizRead
|
||||
from app.services.notiz import erstelle_notiz, liste_fuer_entitaet, loesche_notiz
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
_mitarbeiter_plus = require_roles(
|
||||
RolleTyp.mitarbeiter,
|
||||
RolleTyp.materialverantwortlicher,
|
||||
RolleTyp.leitungsverantwortlicher,
|
||||
RolleTyp.administration,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/notizen", response_model=list[AkteNotizRead])
|
||||
async def liste_notizen(
|
||||
entitaet_typ: EntitaetTyp,
|
||||
entitaet_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
_=Depends(_mitarbeiter_plus),
|
||||
) -> list[AkteNotiz]:
|
||||
return await liste_fuer_entitaet(db, entitaet_typ=entitaet_typ, entitaet_id=entitaet_id)
|
||||
|
||||
|
||||
@router.post("/notizen", response_model=AkteNotizRead, status_code=status.HTTP_201_CREATED)
|
||||
async def notiz_anlegen(
|
||||
payload: AkteNotizCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: Benutzer = Depends(_mitarbeiter_plus),
|
||||
) -> AkteNotiz:
|
||||
return await erstelle_notiz(
|
||||
db,
|
||||
entitaet_typ=payload.entitaet_typ,
|
||||
entitaet_id=payload.entitaet_id,
|
||||
text=payload.text,
|
||||
autor_id=current_user.id,
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/notizen/{notiz_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def notiz_loeschen(
|
||||
notiz_id: uuid.UUID,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: Benutzer = Depends(_mitarbeiter_plus),
|
||||
) -> None:
|
||||
notiz = await db.get(AkteNotiz, notiz_id)
|
||||
if notiz is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notiz nicht gefunden")
|
||||
ist_administrator = RolleTyp.administration.value in current_user.rollen_namen
|
||||
if notiz.autor_id != current_user.id and not ist_administrator:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN, detail="Nur eigene Notizen oder als Administrator löschbar"
|
||||
)
|
||||
await loesche_notiz(db, notiz=notiz)
|
||||
@@ -12,6 +12,7 @@ from app.models.lagerbewegung import Lagerbewegung
|
||||
from app.models.mangel import Mangel, MangelPrioritaet, MangelStatus
|
||||
from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
|
||||
from app.models.nachfuellung import Nachfuellung
|
||||
from app.models.notiz import AkteNotiz
|
||||
from app.models.objekt import Objekt, ObjektStatus
|
||||
from app.models.objektposition import Objektposition, ObjektpositionStatus
|
||||
from app.models.permission import Berechtigung, BenutzerRolleZuordnung, Rolle, RolleBerechtigung
|
||||
@@ -65,6 +66,7 @@ __all__ = [
|
||||
"MindermengeStatus",
|
||||
"MindermengenGenehmigung",
|
||||
"Nachfuellung",
|
||||
"AkteNotiz",
|
||||
"Objekt",
|
||||
"ObjektStatus",
|
||||
"Objektposition",
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import ForeignKey, String
|
||||
from sqlalchemy.dialects.postgresql import TIMESTAMP, UUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class AkteNotiz(Base):
|
||||
"""FILE-009: formlose Freitext-Notiz an beliebiger Ressource - gleiches
|
||||
entitaet_typ/entitaet_id-Muster wie Dokument/Historie/Mangel."""
|
||||
|
||||
__tablename__ = "akte_notiz"
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
entitaet_typ: Mapped[str] = mapped_column(String, nullable=False)
|
||||
entitaet_id: Mapped[str] = mapped_column(String, nullable=False)
|
||||
text: Mapped[str] = mapped_column(String, nullable=False)
|
||||
autor_id: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
|
||||
erstellt_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
|
||||
@@ -0,0 +1,22 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from app.schemas.dokument import EntitaetTyp
|
||||
|
||||
|
||||
class AkteNotizCreate(BaseModel):
|
||||
entitaet_typ: EntitaetTyp
|
||||
entitaet_id: str
|
||||
text: str
|
||||
|
||||
|
||||
class AkteNotizRead(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: uuid.UUID
|
||||
entitaet_typ: str
|
||||
entitaet_id: str
|
||||
text: str
|
||||
autor_id: int
|
||||
erstellt_am: datetime
|
||||
@@ -0,0 +1,38 @@
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.notiz import AkteNotiz
|
||||
|
||||
|
||||
async def erstelle_notiz(
|
||||
db: AsyncSession, *, entitaet_typ: str, entitaet_id: str, text: str, autor_id: int
|
||||
) -> AkteNotiz:
|
||||
notiz = AkteNotiz(
|
||||
id=uuid.uuid4(),
|
||||
entitaet_typ=entitaet_typ,
|
||||
entitaet_id=entitaet_id,
|
||||
text=text,
|
||||
autor_id=autor_id,
|
||||
erstellt_am=datetime.now(timezone.utc),
|
||||
)
|
||||
db.add(notiz)
|
||||
await db.commit()
|
||||
await db.refresh(notiz)
|
||||
return notiz
|
||||
|
||||
|
||||
async def liste_fuer_entitaet(db: AsyncSession, *, entitaet_typ: str, entitaet_id: str) -> list[AkteNotiz]:
|
||||
ergebnis = await db.execute(
|
||||
select(AkteNotiz)
|
||||
.where(AkteNotiz.entitaet_typ == entitaet_typ, AkteNotiz.entitaet_id == entitaet_id)
|
||||
.order_by(AkteNotiz.erstellt_am.desc())
|
||||
)
|
||||
return list(ergebnis.scalars().all())
|
||||
|
||||
|
||||
async def loesche_notiz(db: AsyncSession, *, notiz: AkteNotiz) -> None:
|
||||
await db.delete(notiz)
|
||||
await db.commit()
|
||||
@@ -0,0 +1,60 @@
|
||||
import pytest
|
||||
|
||||
from tests.conftest import auth_header, login
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notiz_anlegen_und_lesen(client, objekt_mit_position, mitarbeiter_user):
|
||||
objekt, _material = objekt_mit_position
|
||||
token = await login(client, "mitarbeiter1")
|
||||
|
||||
angelegt = await client.post(
|
||||
"/api/v1/notizen",
|
||||
json={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "text": "Funkgerät klemmt manchmal"},
|
||||
headers=auth_header(token),
|
||||
)
|
||||
assert angelegt.status_code == 201
|
||||
|
||||
zweite = await client.post(
|
||||
"/api/v1/notizen",
|
||||
json={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "text": "Rücksprache Hersteller 12.03."},
|
||||
headers=auth_header(token),
|
||||
)
|
||||
assert zweite.status_code == 201
|
||||
|
||||
liste = await client.get(
|
||||
f"/api/v1/notizen?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(token)
|
||||
)
|
||||
assert liste.status_code == 200
|
||||
daten = liste.json()
|
||||
assert len(daten) == 2
|
||||
# neueste zuerst
|
||||
assert daten[0]["text"] == "Rücksprache Hersteller 12.03."
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fremde_notiz_nicht_loeschbar_ausser_admin(client, objekt_mit_position, mitarbeiter_user, admin_user):
|
||||
objekt, _material = objekt_mit_position
|
||||
token_mitarbeiter = await login(client, "mitarbeiter1")
|
||||
|
||||
angelegt = await client.post(
|
||||
"/api/v1/notizen",
|
||||
json={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "text": "Testnotiz"},
|
||||
headers=auth_header(token_mitarbeiter),
|
||||
)
|
||||
notiz_id = angelegt.json()["id"]
|
||||
|
||||
token_admin = await login(client, "admin1")
|
||||
abgelehnt = await client.delete(f"/api/v1/notizen/{notiz_id}", headers=auth_header(token_mitarbeiter))
|
||||
# eigene Notiz darf der Autor selbst löschen
|
||||
assert abgelehnt.status_code == 204
|
||||
|
||||
zweite = await client.post(
|
||||
"/api/v1/notizen",
|
||||
json={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "text": "Zweite Notiz"},
|
||||
headers=auth_header(token_mitarbeiter),
|
||||
)
|
||||
zweite_id = zweite.json()["id"]
|
||||
|
||||
erlaubt = await client.delete(f"/api/v1/notizen/{zweite_id}", headers=auth_header(token_admin))
|
||||
assert erlaubt.status_code == 204
|
||||
Reference in New Issue
Block a user