patrickandClaude Sonnet 5
23f8e634d1
GET /users/{id}/export-data liefert alle personenbezogenen Daten
(Profil, Zeiterfassung, Abwesenheiten, Kommentare, Urlaubs-/Überstunden-
konto, Auszahlungen, AuditLog-Einträge über den User) als JSON-Download.
Eigenes Konto immer erlaubt, fremde Konten nur HR/Admin/Manager der
eigenen Firma. Ergänzt die bestehende Art.-17-Anonymisierung.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CviFgc739S89xS97pvjszj
128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
"""DSGVO Art. 15 (Auskunftsrecht): strukturierter Export aller personenbezogenen
|
|
Daten eines Users. Rein lesend, keine Schreiboperation. Ergänzt die bereits
|
|
vorhandene Art.-17-Anonymisierung (user_service.anonymize_user)."""
|
|
import json
|
|
from datetime import date, datetime, time
|
|
from decimal import Decimal
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.absence import Absence
|
|
from app.models.absence_comment import AbsenceComment
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.hours_payout import HoursPayout
|
|
from app.models.overtime_balance import OvertimeBalance
|
|
from app.models.time_entry import TimeEntry
|
|
from app.models.user import User
|
|
from app.models.vacation_balance import VacationBalance
|
|
|
|
|
|
def _json_default(value):
|
|
if isinstance(value, (datetime, date, time)):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
if isinstance(value, UUID):
|
|
return str(value)
|
|
raise TypeError(f"Nicht serialisierbar: {type(value)}")
|
|
|
|
|
|
def _row(obj, fields: list[str]) -> dict:
|
|
return {f: getattr(obj, f) for f in fields}
|
|
|
|
|
|
class GdprService:
|
|
|
|
async def export_user_data(self, user_id: UUID, company_id: UUID, db: AsyncSession) -> dict:
|
|
"""Alle personenbezogenen Daten eines Users, company-isoliert. Wirft
|
|
ValueError wenn der User nicht in der Firma existiert (404 im Router)."""
|
|
user = await db.get(User, user_id)
|
|
if user is None or user.company_id != company_id:
|
|
raise ValueError("Mitarbeiter nicht gefunden")
|
|
|
|
profile = _row(user, [
|
|
"id", "first_name", "last_name", "email", "role", "personnel_number",
|
|
"is_active", "entry_date", "exit_date", "created_at",
|
|
])
|
|
|
|
time_entries = [
|
|
_row(e, [
|
|
"id", "date", "start_time", "end_time", "break_minutes",
|
|
"worked_hours", "status", "source", "note", "created_at",
|
|
])
|
|
for e in (await db.scalars(
|
|
select(TimeEntry).where(TimeEntry.user_id == user_id).order_by(TimeEntry.date)
|
|
)).all()
|
|
]
|
|
|
|
absences = [
|
|
_row(a, [
|
|
"id", "type_id", "start_date", "end_date", "half_day_start",
|
|
"half_day_end", "working_days", "status", "note",
|
|
"rejection_reason", "substitute_id", "created_at",
|
|
])
|
|
for a in (await db.scalars(
|
|
select(Absence).where(Absence.user_id == user_id).order_by(Absence.start_date)
|
|
)).all()
|
|
]
|
|
|
|
absence_comments = [
|
|
_row(c, ["id", "absence_id", "body", "is_system", "created_at"])
|
|
for c in (await db.scalars(
|
|
select(AbsenceComment).where(AbsenceComment.author_id == user_id)
|
|
)).all()
|
|
]
|
|
|
|
vacation_balances = [
|
|
_row(v, ["id", "year", "entitled_days", "special_days", "carried_over", "used_days"])
|
|
for v in (await db.scalars(
|
|
select(VacationBalance).where(VacationBalance.user_id == user_id).order_by(VacationBalance.year)
|
|
)).all()
|
|
]
|
|
|
|
overtime_balance = await db.scalar(
|
|
select(OvertimeBalance).where(OvertimeBalance.user_id == user_id)
|
|
)
|
|
overtime = _row(overtime_balance, ["total_hours", "taken_hours"]) if overtime_balance else None
|
|
|
|
hours_payouts = [
|
|
_row(p, [
|
|
"id", "hours", "period_year", "period_month", "note", "status",
|
|
"rejection_reason", "decided_at", "created_at",
|
|
])
|
|
for p in (await db.scalars(
|
|
select(HoursPayout).where(HoursPayout.user_id == user_id).order_by(HoursPayout.created_at)
|
|
)).all()
|
|
]
|
|
|
|
# AuditLog: nur Einträge über die Person, keine, bei denen sie nur Akteur war
|
|
# (letzteres wäre kein Datum ÜBER die Person, sondern über deren Handlung an Dritten)
|
|
audit_entries = [
|
|
_row(a, ["id", "action", "entity_type", "entity_id", "created_at"])
|
|
for a in (await db.scalars(
|
|
select(AuditLog).where(
|
|
AuditLog.entity_type == "user", AuditLog.entity_id == user_id
|
|
).order_by(AuditLog.created_at)
|
|
)).all()
|
|
]
|
|
|
|
return {
|
|
"export_created_at": datetime.utcnow(),
|
|
"profile": profile,
|
|
"time_entries": time_entries,
|
|
"absences": absences,
|
|
"absence_comments": absence_comments,
|
|
"vacation_balances": vacation_balances,
|
|
"overtime_balance": overtime,
|
|
"hours_payouts": hours_payouts,
|
|
"audit_log_about_user": audit_entries,
|
|
}
|
|
|
|
def to_json(self, data: dict) -> str:
|
|
return json.dumps(data, default=_json_default, indent=2, ensure_ascii=False)
|
|
|
|
|
|
gdpr_service = GdprService()
|