feat(reports): DATEV-Monatsblatt-Export (Blanko-Vorlage) für Steuerberater
Security Audit / Python Dependency Audit (push) Has been cancelled
Security Audit / Node.js Dependency Audit (push) Has been cancelled

Neuer PDF-Export GET /reports/datev-monthly/export: ein Blatt pro
Mitarbeiter/Monat im Layout der DATEV-Stundenaufzeichnungs-Vorlage
(Beginn/Pause/Ende/Dauer + K/U/UU/F/SA/SU-Kürzel + Bemerkungen +
Summe + Unterschriftsfelder). Führt time_entries, Absences und
Feiertage pro Kalendertag zusammen.

Außerdem: Mustervorlagen für neue Backend-Module (Model/Schema/Router/
Migration/Test) und eine Frontend-Page-Vorlage, abgeleitet vom
hours_payouts-Modul als aktuellstem sauberen Muster.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CviFgc739S89xS97pvjszj
This commit is contained in:
2026-08-06 22:19:54 +02:00
co-authored by Claude Sonnet 5
parent 90f45a4f85
commit 5cbeebe66d
10 changed files with 901 additions and 2 deletions
+23 -1
View File
@@ -1,7 +1,7 @@
from datetime import date, timedelta from datetime import date, timedelta
from uuid import UUID from uuid import UUID
from fastapi import APIRouter, Depends, Query from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import Response from fastapi.responses import Response
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@@ -152,6 +152,28 @@ async def export_time_report(
headers={"Content-Disposition": f"attachment; filename={filename}.csv"}) headers={"Content-Disposition": f"attachment; filename={filename}.csv"})
@router.get("/reports/datev-monthly/export")
async def export_datev_monthly(
current_user: CurrentUser,
user_id: UUID,
year: int = Query(..., ge=2000, le=2100),
month: int = Query(..., ge=1, le=12),
db: AsyncSession = Depends(get_db),
):
"""DATEV-Monatsblatt (Blanko-Layout) als PDF für Steuerberater-Übergabe.
EMPLOYEE darf nur den eigenen Monat exportieren."""
if current_user.role not in _manager_roles and user_id != current_user.id:
raise HTTPException(403, "Nur eigenes Monatsblatt exportierbar")
sheet = await report_service.datev_monthly_report(
current_user.company_id, user_id, year, month, db
)
content = report_service.datev_monthly_report_to_pdf(sheet)
filename = f"datev_stundenaufzeichnung_{sheet.user_name.replace(' ', '_')}_{year}_{month:02d}.pdf"
return Response(content=content, media_type="application/pdf",
headers={"Content-Disposition": f"attachment; filename={filename}"})
@router.get("/reports/absences/export") @router.get("/reports/absences/export")
async def export_absence_report( async def export_absence_report(
current_user: CurrentUser, current_user: CurrentUser,
+26
View File
@@ -188,3 +188,29 @@ class OvertimeReportDetailed(BaseModel):
total_employees: int total_employees: int
total_overtime: float total_overtime: float
rows: list[OvertimeReportRowDetailed] rows: list[OvertimeReportRowDetailed]
# ── DATEV-Monatsblatt ──────────────────────────────────────────────────────────
# Layout entspricht "Stundenaufzeichnungen Muster blanko DATEV ab 2015":
# ein Blatt pro Mitarbeiter/Monat, Zeile pro Kalendertag.
class DatevDayRow(BaseModel):
day: int # Kalendertag 1-31
weekday_label: str # "Mo".."So" für Anzeige
start_time: time | None = None
break_minutes: int | None = None
end_time: time | None = None
duration_hours: float | None = None
code: str | None = None # K/U/UU/F/SA/SU siehe DATEV_CODES
recorded_on: date | None = None # "aufgezeichnet am" (created_at des Eintrags)
note: str | None = None # Bemerkungen (Absence-Typ-Name o.ä.)
class DatevMonthlySheet(BaseModel):
company_name: str
user_name: str
personnel_number: str | None
year: int
month: int
rows: list[DatevDayRow]
total_hours: float
+193 -1
View File
@@ -1,3 +1,4 @@
import calendar
import csv import csv
import io import io
from collections import defaultdict from collections import defaultdict
@@ -9,7 +10,7 @@ from sqlalchemy import distinct, func, select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.models.absence import Absence, AbsenceStatus from app.models.absence import Absence, AbsenceStatus
from app.models.absence_type import AbsenceType from app.models.absence_type import AbsenceCategory, AbsenceType
from app.models.company import Company from app.models.company import Company
from app.models.department import Department from app.models.department import Department
from app.models.overtime_balance import OvertimeBalance from app.models.overtime_balance import OvertimeBalance
@@ -21,6 +22,8 @@ from app.schemas.report import (
AbsenceReport, AbsenceReport,
AbsenceReportRow, AbsenceReportRow,
CompanyDashboard, CompanyDashboard,
DatevDayRow,
DatevMonthlySheet,
DayEntry, DayEntry,
EmployeeDashboard, EmployeeDashboard,
HoursBreakdown, HoursBreakdown,
@@ -675,6 +678,195 @@ class ReportService:
total_rows=len(rows), total_hours=round(total_hours, 2), rows=rows, total_rows=len(rows), total_hours=round(total_hours, 2), rows=rows,
) )
# ── DATEV-Monatsblatt ────────────────────────────────────────────────────
# Kürzel-Mapping laut Vorlage "Stundenaufzeichnungen Muster blanko DATEV ab 2015":
# K=Krank, U=Urlaub, UU=unbezahlter Urlaub, F=Feiertag, SA=Stundenweise abwesend,
# SU=Stundenweise Urlaub.
_WEEKDAY_LABELS = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
@staticmethod
def _datev_code_for_absence(absence_type: AbsenceType, is_half_day: bool) -> str:
if absence_type.category == AbsenceCategory.SICK:
return "K"
if absence_type.category == AbsenceCategory.VACATION:
if is_half_day:
return "SU"
return "U" if absence_type.is_paid else "UU"
# Alle übrigen Kategorien (overtime_comp, training, business_trip, other)
# haben kein eigenes DATEV-Kürzel -> nur Bemerkungen-Spalte
return "SA" if is_half_day else ""
async def datev_monthly_report(
self,
company_id: UUID,
user_id: UUID,
year: int,
month: int,
db: AsyncSession,
) -> DatevMonthlySheet:
"""Ein-Seite-pro-Mitarbeiter-Monatsblatt im DATEV-Blanko-Layout."""
user = await db.get(User, user_id)
if user is None or user.company_id != company_id:
raise ValueError("Mitarbeiter nicht gefunden")
company = await db.get(Company, company_id)
days_in_month = calendar.monthrange(year, month)[1]
date_from = date(year, month, 1)
date_to = date(year, month, days_in_month)
# Zeiteinträge des Monats
entries_stmt = select(TimeEntry).where(
TimeEntry.user_id == user_id,
TimeEntry.date >= date_from,
TimeEntry.date <= date_to,
)
entries_by_day: dict[int, TimeEntry] = {
e.date.day: e for e in (await db.scalars(entries_stmt)).all()
}
# Genehmigte Abwesenheiten, die den Monat überlappen
absences_stmt = (
select(Absence, AbsenceType)
.join(AbsenceType, Absence.absence_type_id == AbsenceType.id)
.where(
Absence.user_id == user_id,
Absence.status.in_([AbsenceStatus.APPROVED, AbsenceStatus.FIRST_APPROVED]),
Absence.start_date <= date_to,
Absence.end_date >= date_from,
)
)
absences = list((await db.execute(absences_stmt)).all())
# Feiertage (nur wenn Bundesland konfiguriert)
holidays: dict[date, tuple[str, bool]] = {}
if company and company.state:
holidays = await get_holidays_set(date_from, date_to, company.state, db)
rows: list[DatevDayRow] = []
total_hours = 0.0
for day in range(1, days_in_month + 1):
d = date(year, month, day)
weekday_label = self._WEEKDAY_LABELS[d.weekday()]
row = DatevDayRow(day=day, weekday_label=weekday_label)
entry = entries_by_day.get(day)
if entry is not None:
row.start_time = entry.start_time
row.end_time = entry.end_time
row.break_minutes = entry.break_minutes
row.duration_hours = entry.worked_hours
row.recorded_on = entry.created_at.date() if entry.created_at else None
if entry.worked_hours:
total_hours += entry.worked_hours
# Feiertag hat Vorrang vor Abwesenheit (analog _categorize_hours-Logik)
if d in holidays:
row.code = "F"
row.note = holidays[d][0]
else:
for absence, absence_type in absences:
if absence.start_date <= d <= absence.end_date:
is_half_day = (
(d == absence.start_date and absence.half_day_start)
or (d == absence.end_date and absence.half_day_end)
)
row.code = self._datev_code_for_absence(absence_type, is_half_day)
row.note = absence_type.name
break
rows.append(row)
return DatevMonthlySheet(
company_name=company.name if company else "",
user_name=f"{user.first_name} {user.last_name}",
personnel_number=user.personnel_number,
year=year, month=month,
rows=rows, total_hours=round(total_hours, 2),
)
def datev_monthly_report_to_pdf(self, sheet: DatevMonthlySheet) -> bytes:
def fmt_t(t: time | None) -> str:
return t.strftime("%H:%M") if t else ""
def fmt_h(h: float | None) -> str:
return f"{h:.2f}".replace(".", ",") if h is not None else ""
rows_html = ""
for r in sheet.rows:
break_str = f"{r.break_minutes} min" if r.break_minutes else ""
recorded = r.recorded_on.strftime("%d.%m.%Y") if r.recorded_on else ""
rows_html += f"""<tr>
<td>{r.day}. ({r.weekday_label})</td>
<td>{fmt_t(r.start_time)}</td>
<td>{break_str}</td>
<td>{fmt_t(r.end_time)}</td>
<td class="right">{fmt_h(r.duration_hours)}</td>
<td class="bold" style="text-align:center">{r.code or ""}</td>
<td>{recorded}</td>
<td>{r.note or ""}</td>
</tr>"""
month_names = [
"Januar", "Februar", "März", "April", "Mai", "Juni",
"Juli", "August", "September", "Oktober", "November", "Dezember",
]
period = f"{month_names[sheet.month - 1]} {sheet.year}"
html = f"""<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="utf-8">
<style>
@page {{ margin: 1.2cm 1.5cm; }}
body {{ font-family: Arial, Helvetica, sans-serif; font-size: 9px; color: #111827; }}
h1 {{ font-size: 14px; margin: 0 0 10px; }}
.meta {{ display: flex; gap: 30px; margin-bottom: 12px; font-size: 10px; }}
.meta div span {{ display: block; color: #6b7280; font-size: 8px; text-transform: uppercase; }}
table {{ width: 100%; border-collapse: collapse; }}
thead th {{ background: #1e40af; color: white; padding: 4px 6px; text-align: left; font-size: 8px; text-transform: uppercase; }}
td {{ padding: 3px 6px; border-bottom: 1px solid #e5e7eb; }}
td.right {{ text-align: right; }}
td.bold {{ font-weight: 700; }}
tfoot td {{ background: #1e293b; color: white; font-weight: bold; padding: 5px 6px; }}
.signatures {{ display: flex; justify-content: space-between; margin-top: 40px; }}
.sig-line {{ width: 45%; border-top: 1px solid #111827; padding-top: 4px; font-size: 9px; text-align: center; }}
.codes {{ margin-top: 16px; font-size: 8px; color: #4b5563; }}
</style>
</head>
<body>
<h1>Stundenaufzeichnung {period}</h1>
<div class="meta">
<div><span>Firma</span>{sheet.company_name}</div>
<div><span>Name des Mitarbeiters</span>{sheet.user_name}</div>
<div><span>Pers.-Nr.</span>{sheet.personnel_number or ""}</div>
</div>
<table>
<thead><tr>
<th>Kalendertag</th><th>Beginn</th><th>Pause</th><th>Ende</th>
<th>Dauer</th><th>*</th><th>aufgezeichnet am</th><th>Bemerkungen</th>
</tr></thead>
<tbody>
{rows_html}
</tbody>
<tfoot><tr>
<td colspan="4">Summe</td>
<td class="right">{fmt_h(sheet.total_hours)}</td>
<td colspan="3"></td>
</tr></tfoot>
</table>
<div class="codes">* K=Krank · U=Urlaub · UU=unbezahlter Urlaub · F=Feiertag · SA=Stundenweise abwesend · SU=Stundenweise Urlaub</div>
<div class="signatures">
<div class="sig-line">Datum, Unterschrift Arbeitnehmer</div>
<div class="sig-line">Datum, Unterschrift Arbeitgeber</div>
</div>
</body>
</html>"""
from weasyprint import HTML
return HTML(string=html).write_pdf()
# ── Absence Report ─────────────────────────────────────────────────────── # ── Absence Report ───────────────────────────────────────────────────────
async def absence_report( async def absence_report(
+17
View File
@@ -0,0 +1,17 @@
# Mustervorlagen (Backend)
Kopiervorlagen für ein neues Workflow-Modul (Model + Schema + Router + Migration + Test),
abgeleitet vom aktuellsten sauberen Muster im Projekt: `hours_payouts` (Migration 0042).
Reihenfolge beim Bau eines neuen Moduls:
1. `model_template.py``app/models/<name>.py`
2. `schema_template.py``app/schemas/<name>.py`
3. `router_template.py``app/routers/<name>.py`, in `app/main.py` registrieren
4. `migration_template.py``migrations/versions/00XX_<name>.py`, `conftest.py` RLS nachziehen
5. `test_template.py``tests/test_<name>.py`
Jede Datei hat einen Anleitung-Docstring am Kopf. Kern-Fallstricke sind dort verlinkt
(RLS-Bypass + mid-request commit, Company-Isolation, conftest.py-Replikation).
Diese Ordner werden NICHT von der App importiert (kein `__init__.py`) reine Vorlagen.
+69
View File
@@ -0,0 +1,69 @@
"""TEMPLATE Mustervorlage für Migration mit NEUER Tabelle + RLS-Policy.
Abgeleitet vom Muster in migrations/versions/0035_absence_comments.py.
WICHTIG (siehe project_dsgvo_multitenant_priority):
- JEDE neue company-bezogene Tabelle braucht company_id-Spalte + RLS-Policy
- NACH dieser Migration: tests/conftest.py muss die RLS-Policy manuell
replizieren (Test-DB nutzt keine echten Migrationen) sonst schlagen
Company-Isolation-Tests unbemerkt fehl (false green)
- Nur neue SPALTEN auf bereits RLS-gefencten Tabellen brauchen KEINE
RLS-Änderung (siehe 0042_payout_requests.py als Gegenbeispiel)
Anleitung:
1. Nach backend/migrations/versions/00XX_xxx_things.py kopieren
2. revision/down_revision auf nächste freie Nummer setzen (siehe
"Datenbank-Migrationen Chronik" in docmost für die aktuell höchste Nummer)
3. Tabellenname/Spalten anpassen
4. conftest.py RLS-Replikation ergänzen
5. Lokal (auf 137!) testen: alembic upgrade head, dann pytest
"""
from alembic import op
from sqlalchemy import text
revision = "00XX" # TODO: nächste freie Nummer
down_revision = "00XX-1" # TODO: vorherige Migration
branch_labels = None
depends_on = None
_BYPASS = "COALESCE(current_setting('app.bypass_rls', true), 'off') = 'on'"
_CID = "company_id = NULLIF(current_setting('app.company_id', true), '')::uuid"
_USING = f"({_BYPASS} OR {_CID})"
def _exec(sql: str) -> None:
op.execute(text(sql))
def upgrade() -> None:
# 1) Tabelle anlegen
_exec("""
CREATE TABLE IF NOT EXISTS xxx_things (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
company_id UUID NOT NULL REFERENCES companies(id) ON DELETE CASCADE,
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
note TEXT,
status VARCHAR(12) NOT NULL DEFAULT 'requested',
rejection_reason TEXT,
created_by UUID NOT NULL REFERENCES users(id) ON DELETE SET NULL,
decided_by UUID REFERENCES users(id) ON DELETE SET NULL,
decided_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
""")
_exec("CREATE INDEX IF NOT EXISTS ix_xxx_things_company_id ON xxx_things(company_id)")
_exec("CREATE INDEX IF NOT EXISTS ix_xxx_things_user_id ON xxx_things(user_id)")
# 2) RLS (company_id-gefenced)
_exec("ALTER TABLE xxx_things ENABLE ROW LEVEL SECURITY")
_exec("ALTER TABLE xxx_things FORCE ROW LEVEL SECURITY")
for cmd in ("select", "insert", "update", "delete"):
_exec(f"DROP POLICY IF EXISTS rls_xxx_things_{cmd} ON xxx_things")
_exec(f"CREATE POLICY rls_xxx_things_select ON xxx_things FOR SELECT USING {_USING}")
_exec(f"CREATE POLICY rls_xxx_things_insert ON xxx_things FOR INSERT WITH CHECK {_USING}")
_exec(f"CREATE POLICY rls_xxx_things_update ON xxx_things FOR UPDATE USING {_USING} WITH CHECK {_USING}")
_exec(f"CREATE POLICY rls_xxx_things_delete ON xxx_things FOR DELETE USING {_USING}")
def downgrade() -> None:
_exec("DROP TABLE IF EXISTS xxx_things")
+67
View File
@@ -0,0 +1,67 @@
"""TEMPLATE Mustervorlage für ein neues Model.
Anleitung:
1. Datei nach app/models/<name>.py kopieren
2. XxxThing, xxx_things, xxx_thing_id konsequent ersetzen
3. Enum/Felder an das reale Fachmodell anpassen
4. In app/models/__init__.py importieren (Alembic autodiscovery)
5. RLS-Policy in der zugehörigen Migration NICHT vergessen (company_id-Fenced)
siehe conftest.py, dort muss die Policy für Tests manuell repliziert werden.
"""
import enum
import uuid
from datetime import datetime
from typing import TYPE_CHECKING
from sqlalchemy import DateTime, ForeignKey, String, Text, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
if TYPE_CHECKING:
from app.models.user import User
from app.models.company import Company
class XxxThingStatus(str, enum.Enum):
REQUESTED = "requested"
APPROVED = "approved"
REJECTED = "rejected"
CANCELLED = "cancelled"
class XxxThing(Base):
"""Kurzbeschreibung was dieses Model fachlich abbildet."""
__tablename__ = "xxx_things"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
# Mandanten-Fenced Pflicht für RLS-Isolation (DSGVO)
company_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"),
nullable=False, index=True
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"),
nullable=False, index=True
)
note: Mapped[str | None] = mapped_column(Text)
status: Mapped[str] = mapped_column(String(12), nullable=False, default=XxxThingStatus.REQUESTED.value)
created_by: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=False
)
decided_by: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL")
)
decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True
)
user: Mapped["User"] = relationship("User", foreign_keys=[user_id], lazy="noload")
creator: Mapped["User"] = relationship("User", foreign_keys=[created_by], lazy="noload")
decider: Mapped["User"] = relationship("User", foreign_keys=[decided_by], lazy="noload")
company: Mapped["Company"] = relationship("Company", lazy="noload")
+168
View File
@@ -0,0 +1,168 @@
"""TEMPLATE Mustervorlage für einen neuen Router (Modul mit Workflow-Status).
Abgeleitet vom aktuellsten sauberen Muster im Projekt: app/routers/hours_payouts.py.
Kern-Regeln (aus CLAUDE.md + gelernte Fallstricke):
- Company-Isolation: JEDE Query filtert auf company_id == current_user.company_id
(zusätzlich zur DB-seitigen RLS Python-Check ist die zweite Verteidigungslinie)
- **RLS-Bypass + mid-request commit Falle**: Response-Objekt (_build_out) IMMER
VOR db.commit() bauen. db.refresh() nur nach db.flush() (noch in Transaktion).
Nach commit() ist app.bypass_rls/app.company_id verfallen -> Post-Commit-Reads
sehen ggf. keine Zeilen mehr.
- AuditLog bei JEDER schreibenden Aktion (create/approve/reject/cancel/delete)
- HTTPException mit sprechendem detail, kein Hard-Delete
- require_role() aus core.dependencies für Rollenprüfung
Anleitung:
1. Nach app/routers/<name>.py kopieren, XxxThing/xxx_things ersetzen
2. In app/main.py registrieren: app.include_router(xxx_things.router, prefix="/api/v1")
3. Model + Schema aus model_template.py / schema_template.py zuerst anlegen
4. Migration schreiben (Alembic) inkl. RLS-Policy + conftest.py nachziehen
5. Tests in backend/tests/test_xxx_things.py (siehe test_hours_payouts.py als Muster)
"""
from datetime import datetime
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.dependencies import get_client_ip, require_role
from app.models.audit_log import AuditLog
from app.models.user import User, UserRole
from app.models.xxx_thing import XxxThing, XxxThingStatus # TODO: Model anlegen
from app.schemas.xxx_thing import ( # TODO: Schemas anlegen
XxxThingCreate,
XxxThingListResponse,
XxxThingOut,
XxxThingReject,
)
router = APIRouter(tags=["Xxx-Thing"])
_hr_roles = (UserRole.HR, UserRole.COMPANY_ADMIN, UserRole.SUPER_ADMIN)
_all_roles = (UserRole.EMPLOYEE, UserRole.MANAGER, UserRole.HR, UserRole.COMPANY_ADMIN, UserRole.SUPER_ADMIN)
async def _build_out(item: XxxThing, db: AsyncSession) -> XxxThingOut:
"""Response-Objekt VOR commit() bauen siehe RLS-Bypass-Falle oben."""
out = XxxThingOut.model_validate(item)
user = await db.get(User, item.user_id)
out.user_name = f"{user.first_name} {user.last_name}" if user else str(item.user_id)
return out
# ── GET /xxx-things ────────────────────────────────────────────────────────
@router.get("/xxx-things", response_model=XxxThingListResponse)
async def list_items(
status: str | None = Query(None),
current_user: User = require_role(*_all_roles),
db: AsyncSession = Depends(get_db),
):
"""Liste der eigenen Firma. EMPLOYEE/MANAGER sehen nur eigene Einträge."""
filters = [XxxThing.company_id == current_user.company_id]
if current_user.role not in _hr_roles:
filters.append(XxxThing.user_id == current_user.id)
if status is not None:
filters.append(XxxThing.status == status)
total_count = await db.scalar(select(func.count()).select_from(XxxThing).where(*filters))
rows = list(await db.scalars(
select(XxxThing).where(*filters).order_by(XxxThing.created_at.desc())
))
result = [await _build_out(row, db) for row in rows]
return XxxThingListResponse(items=result, total_count=total_count or 0)
# ── POST /xxx-things ───────────────────────────────────────────────────────
@router.post("/xxx-things", response_model=XxxThingOut, status_code=201)
async def create_item(
request: Request,
data: XxxThingCreate,
current_user: User = require_role(*_all_roles),
db: AsyncSession = Depends(get_db),
):
item = XxxThing(
company_id=current_user.company_id,
user_id=data.user_id,
note=data.note,
status=XxxThingStatus.REQUESTED.value,
created_by=current_user.id,
)
db.add(item)
await db.flush()
await db.refresh(item) # server_default created_at noch in-Transaktion
db.add(AuditLog(
company_id=current_user.company_id, user_id=current_user.id,
action="xxx_thing_created", entity_type="xxx_thing", entity_id=item.id,
new_value={"note": data.note},
ip=get_client_ip(request),
))
out = await _build_out(item, db)
await db.commit()
return out
# ── POST /xxx-things/{id}/approve ──────────────────────────────────────────
@router.post("/xxx-things/{item_id}/approve", response_model=XxxThingOut)
async def approve_item(
item_id: UUID,
request: Request,
current_user: User = require_role(*_hr_roles),
db: AsyncSession = Depends(get_db),
):
item = await db.get(XxxThing, item_id)
if item is None or item.company_id != current_user.company_id:
raise HTTPException(404, "Eintrag nicht gefunden")
if item.status != XxxThingStatus.REQUESTED.value:
raise HTTPException(409, "Nur offene Anträge können genehmigt werden.")
item.status = XxxThingStatus.APPROVED.value
item.decided_by = current_user.id
item.decided_at = datetime.utcnow()
db.add(AuditLog(
company_id=current_user.company_id, user_id=current_user.id,
action="xxx_thing_approved", entity_type="xxx_thing", entity_id=item.id,
ip=get_client_ip(request),
))
out = await _build_out(item, db)
await db.commit()
return out
# ── POST /xxx-things/{id}/reject ───────────────────────────────────────────
@router.post("/xxx-things/{item_id}/reject", response_model=XxxThingOut)
async def reject_item(
item_id: UUID,
request: Request,
data: XxxThingReject,
current_user: User = require_role(*_hr_roles),
db: AsyncSession = Depends(get_db),
):
item = await db.get(XxxThing, item_id)
if item is None or item.company_id != current_user.company_id:
raise HTTPException(404, "Eintrag nicht gefunden")
if item.status != XxxThingStatus.REQUESTED.value:
raise HTTPException(409, "Nur offene Anträge können abgelehnt werden.")
item.status = XxxThingStatus.REJECTED.value
item.rejection_reason = data.rejection_reason
item.decided_by = current_user.id
item.decided_at = datetime.utcnow()
db.add(AuditLog(
company_id=current_user.company_id, user_id=current_user.id,
action="xxx_thing_rejected", entity_type="xxx_thing", entity_id=item.id,
new_value={"rejection_reason": data.rejection_reason},
ip=get_client_ip(request),
))
out = await _build_out(item, db)
await db.commit()
return out
+42
View File
@@ -0,0 +1,42 @@
"""TEMPLATE Mustervorlage für Pydantic-v2-Schemas.
Anleitung:
1. Nach app/schemas/<name>.py kopieren, XxxThing ersetzen
2. Computed-Felder (z.B. *_name) im Router nachträglich befüllen, nie im Schema
selbst berechnen (Router = einzige Stelle mit DB-Zugriff für den Response-Bau)
"""
import uuid
from datetime import datetime
from pydantic import BaseModel, Field
class XxxThingCreate(BaseModel):
user_id: uuid.UUID
note: str | None = Field(None, max_length=500)
class XxxThingReject(BaseModel):
rejection_reason: str | None = Field(None, max_length=500)
class XxxThingOut(BaseModel):
model_config = {"from_attributes": True}
id: uuid.UUID
company_id: uuid.UUID
user_id: uuid.UUID
user_name: str = "" # Computed im Router (first_name + last_name)
note: str | None
status: str = "requested"
rejection_reason: str | None = None
created_by: uuid.UUID
created_by_name: str = "" # Computed im Router
decided_by: uuid.UUID | None = None
decided_by_name: str = "" # Computed im Router
decided_at: datetime | None = None
created_at: datetime
class XxxThingListResponse(BaseModel):
items: list[XxxThingOut]
total_count: int
+66
View File
@@ -0,0 +1,66 @@
"""TEMPLATE Mustervorlage für Router-Tests. Abgeleitet von test_hours_payouts.py.
Anleitung:
1. Nach backend/tests/test_xxx_things.py kopieren, Endpunkte/Felder anpassen
2. pytest-Konvention: alle Fixtures scope="session" + loop_scope="session"
(asyncpg + pytest-asyncio 1.x Anforderung siehe project_pytest_asyncio)
3. Ausführen NUR auf dem Server (root@192.168.1.137), nie lokal:
ssh root@192.168.1.137 'cd /opt/timemaster/backend && source venv/bin/activate && python -m pytest tests/test_xxx_things.py -v'
"""
import pytest
import pytest_asyncio
from httpx import AsyncClient
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def xxx_thing_headers(client: AsyncClient):
resp = await client.post("/api/v1/auth/register", json={
"company_name": "XxxThing GmbH",
"first_name": "Test",
"last_name": "User",
"email": "admin@xxxthinggmbh.de",
"password": "Secret123",
})
assert resp.status_code == 201, resp.text
return {"Authorization": f"Bearer {resp.json()['access_token']}"}
@pytest.mark.asyncio(loop_scope="session")
async def test_create_item(client: AsyncClient, xxx_thing_headers):
r = await client.post("/api/v1/xxx-things",
json={"user_id": "...", "note": "Test"}, headers=xxx_thing_headers)
assert r.status_code == 201, r.text
assert r.json()["status"] == "requested"
@pytest.mark.asyncio(loop_scope="session")
async def test_approve_item(client: AsyncClient, xxx_thing_headers):
create = await client.post("/api/v1/xxx-things",
json={"user_id": "...", "note": "Test"}, headers=xxx_thing_headers)
item_id = create.json()["id"]
ap = await client.post(f"/api/v1/xxx-things/{item_id}/approve", json={}, headers=xxx_thing_headers)
assert ap.status_code == 200, ap.text
assert ap.json()["status"] == "approved"
# Doppel-Approve muss scheitern (409)
ap2 = await client.post(f"/api/v1/xxx-things/{item_id}/approve", json={}, headers=xxx_thing_headers)
assert ap2.status_code == 409
@pytest.mark.asyncio(loop_scope="session")
async def test_company_isolation(client: AsyncClient, xxx_thing_headers):
"""Cross-Tenant-Zugriff muss 404 liefern, nicht 200/403 (verrät keine Existenz)."""
other = await client.post("/api/v1/auth/register", json={
"company_name": "Andere Firma GmbH",
"first_name": "Other", "last_name": "Admin",
"email": "admin@andere-firma.de", "password": "Secret123",
})
other_headers = {"Authorization": f"Bearer {other.json()['access_token']}"}
create = await client.post("/api/v1/xxx-things",
json={"user_id": "...", "note": "Geheim"}, headers=xxx_thing_headers)
item_id = create.json()["id"]
r = await client.get("/api/v1/xxx-things", headers=other_headers)
assert all(item["id"] != item_id for item in r.json()["items"])
+230
View File
@@ -0,0 +1,230 @@
// TEMPLATE Mustervorlage für eine neue Seite mit Liste + Modal + Approve/Reject.
// Abgeleitet von pages/HoursPayoutPage.tsx.
//
// Anleitung:
// 1. Nach pages/XxxThingsPage.tsx kopieren, XxxThing/xxx-things ersetzen
// 2. Route in App.tsx registrieren, Nav-Eintrag in components/navConfig.ts
// 3. Konventionen: kein useEffect wenn Event-Handler reicht, Fehler als
// string | null, Lade-Zustand mit <Spinner />, Modal mit <Modal>-Komponente
// 4. Bauen + deployen: npm run build && rsync -az --delete ./dist/ root@192.168.1.137:/opt/timemaster/frontend/dist/
import { useEffect, useState } from 'react'
import { api } from '../api/client'
import { Spinner } from '../components/Spinner'
import { Layout } from '../components/Layout'
import { Modal } from '../components/Modal'
interface XxxThingOut {
id: string
user_name: string
note: string | null
status: string
rejection_reason: string | null
created_at: string
}
interface XxxThingListResponse {
items: XxxThingOut[]
total_count: number
}
interface Me {
first_name: string
last_name: string
role: string
}
const inputClass = 'w-full border border-gray-300 rounded-lg px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-purple-400 focus:border-transparent'
const STATUS_BADGE: Record<string, { label: string; cls: string }> = {
requested: { label: 'Beantragt', cls: 'bg-amber-100 text-amber-700' },
approved: { label: 'Genehmigt', cls: 'bg-green-100 text-green-700' },
rejected: { label: 'Abgelehnt', cls: 'bg-red-100 text-red-700' },
}
function StatusBadge({ status }: { status: string }) {
const b = STATUS_BADGE[status] ?? { label: status, cls: 'bg-gray-100 text-gray-500' }
return <span className={`inline-block px-2 py-0.5 rounded-full text-xs font-medium ${b.cls}`}>{b.label}</span>
}
export function XxxThingsPage() {
const [me, setMe] = useState<Me | null>(null)
const [pageLoading, setPageLoading] = useState(true)
const [pageError, setPageError] = useState<string | null>(null)
const [items, setItems] = useState<XxxThingOut[]>([])
const [tableLoading, setTableLoading] = useState(false)
const [showModal, setShowModal] = useState(false)
const [formNote, setFormNote] = useState('')
const [modalSaving, setModalSaving] = useState(false)
const [modalError, setModalError] = useState<string | null>(null)
useEffect(() => {
async function init() {
try {
const meData = await api.get<Me>('/auth/me')
setMe(meData)
await loadItems()
} catch (e: unknown) {
setPageError(e instanceof Error ? e.message : 'Fehler beim Laden')
} finally {
setPageLoading(false)
}
}
init()
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [])
async function loadItems() {
setTableLoading(true)
try {
const data = await api.get<XxxThingListResponse>('/xxx-things')
setItems(data.items)
} catch (e: unknown) {
setPageError(e instanceof Error ? e.message : 'Fehler beim Laden')
} finally {
setTableLoading(false)
}
}
function openNewModal() {
setFormNote('')
setModalError(null)
setShowModal(true)
}
async function handleSave() {
setModalSaving(true)
setModalError(null)
try {
await api.post<XxxThingOut>('/xxx-things', { note: formNote.trim() || null })
setShowModal(false)
loadItems()
} catch (e: unknown) {
setModalError(e instanceof Error ? e.message : 'Fehler beim Anlegen')
} finally {
setModalSaving(false)
}
}
async function handleApprove(item: XxxThingOut) {
if (!confirm(`Eintrag für ${item.user_name} genehmigen?`)) return
try {
const updated = await api.post<XxxThingOut>(`/xxx-things/${item.id}/approve`, {})
setItems(prev => prev.map(i => (i.id === item.id ? updated : i)))
} catch (e: unknown) {
alert(e instanceof Error ? e.message : 'Fehler beim Genehmigen')
}
}
async function handleReject(item: XxxThingOut) {
const reason = prompt('Grund der Ablehnung (optional):')
if (reason === null) return
try {
const updated = await api.post<XxxThingOut>(`/xxx-things/${item.id}/reject`, {
rejection_reason: reason.trim() || null,
})
setItems(prev => prev.map(i => (i.id === item.id ? updated : i)))
} catch (e: unknown) {
alert(e instanceof Error ? e.message : 'Fehler beim Ablehnen')
}
}
if (pageLoading) return (
<div className='min-h-screen bg-gray-50 flex items-center justify-center'><Spinner /></div>
)
if (pageError) return (
<div className='min-h-screen bg-gray-50 flex items-center justify-center'>
<p className='text-red-600'>{pageError}</p>
</div>
)
return (
<Layout userRole={me?.role ?? ''} userName={`${me?.first_name} ${me?.last_name}`}>
<div className='space-y-6'>
<div className='flex items-center justify-between'>
<div>
<h1 className='text-xl font-bold text-gray-800'>Xxx-Things</h1>
<p className='text-sm text-gray-500 mt-0.5'>Kurzbeschreibung der Seite</p>
</div>
<button
onClick={openNewModal}
className='px-4 py-2 bg-purple-600 text-white text-sm font-semibold rounded-lg hover:bg-purple-700 transition-colors'
>
+ Neu anlegen
</button>
</div>
<div className='bg-white rounded-xl border border-gray-200 shadow-sm overflow-hidden'>
{tableLoading ? (
<div className='flex items-center justify-center py-16'><Spinner /></div>
) : (
<div className='overflow-x-auto'>
<table className='w-full text-sm'>
<thead className='bg-gray-50 text-gray-500 text-xs uppercase'>
<tr>
{['Mitarbeiter', 'Status', 'Notiz', 'Datum', 'Aktion'].map(h => (
<th key={h} className='px-4 py-3 text-left font-medium'>{h}</th>
))}
</tr>
</thead>
<tbody className='divide-y divide-gray-100'>
{items.map(item => (
<tr key={item.id} className='hover:bg-gray-50 transition-colors'>
<td className='px-4 py-3 font-medium text-gray-800'>{item.user_name}</td>
<td className='px-4 py-3'><StatusBadge status={item.status} /></td>
<td className='px-4 py-3 text-gray-500 max-w-xs truncate'>{item.note || '—'}</td>
<td className='px-4 py-3 text-gray-500'>
{new Date(item.created_at).toLocaleDateString('de-DE')}
</td>
<td className='px-4 py-3'>
{item.status === 'requested' && (
<div className='flex gap-2'>
<button onClick={() => handleApprove(item)} className='text-xs text-green-600 hover:underline font-medium'> Genehmigen</button>
<button onClick={() => handleReject(item)} className='text-xs text-red-500 hover:underline'> Ablehnen</button>
</div>
)}
</td>
</tr>
))}
{items.length === 0 && (
<tr><td colSpan={5} className='px-4 py-10 text-center text-gray-400'>Keine Einträge gefunden.</td></tr>
)}
</tbody>
</table>
</div>
)}
</div>
</div>
{showModal && (
<Modal title='Neu anlegen' onClose={() => setShowModal(false)}>
<div className='space-y-4'>
<label className='block'>
<span className='text-xs font-medium text-gray-700'>Notiz</span>
<textarea
value={formNote}
onChange={e => setFormNote(e.target.value)}
maxLength={500}
rows={3}
className={inputClass + ' resize-none'}
/>
</label>
{modalError && (
<div className='text-sm text-red-600 bg-red-50 border border-red-200 rounded px-3 py-2'>{modalError}</div>
)}
<div className='flex justify-end gap-2 pt-2'>
<button onClick={() => setShowModal(false)} className='px-4 py-2 text-sm text-gray-700 border border-gray-300 rounded-lg hover:bg-gray-50'>Abbrechen</button>
<button onClick={handleSave} disabled={modalSaving} className='px-4 py-2 text-sm font-semibold text-white bg-purple-600 rounded-lg hover:bg-purple-700 disabled:opacity-50'>
{modalSaving ? 'Speichere…' : 'Anlegen'}
</button>
</div>
</div>
</Modal>
)}
</Layout>
)
}