Backend: - Modelle + Migration 0002: accounts, account_identifiers, category_groups, categories, transactions, exchange_rates (alle owner-gescoped). - EZB-Kursdienst (90-Tage/Historie) + Backfill-Skript; Umrechnung EUR<->CHF zum Buchungsdatum mit lokalem Cache und manueller Übersteuerung; EUR-Pivot (amount_eur_cents) eingefroren. - Kontostände + Netto-Vermögen (Umrechnung in Hauptwährung). - Router: accounts (inkl. IBAN-Identifikatoren), categories (Gruppen+Kategorien), transactions (Ausgabe/Einnahme/Transfer, cross-currency), rates. - Gruppierte Standard-Kategorien je Nutzer beim Anlegen. - pytest-Suite: Währung, Salden, Cross-Currency-Transfer, Nutzer-Isolation. Frontend: - Seiten Konten, Kategorien, Buchungen; Icon-Chip, Geldbetrag mit Umrechnung, Icon-/Farbwähler, Modal; erweitertes SVG-Icon-Set (Konto-/Kategorie-Icons). - Währungs-/Datumsformatierung (EUR de-DE, CHF de-CH), i18n DE/EN erweitert. - Live-Umrechnungs-Vorschau im Transfer-Formular. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
"""Gemeinsame FastAPI-Abhängigkeiten: aktueller Benutzer (Cookie), Admin-Pflicht.
|
||
|
||
Die Anmeldung läuft über eine server-seitige Sitzung im Cookie (siehe
|
||
``services/sessions``). ``get_current_user`` liest das Cookie, prüft die Sitzung
|
||
und liefert den Benutzer – oder 401.
|
||
"""
|
||
|
||
from fastapi import Depends, HTTPException, Request, status
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .config import get_settings
|
||
from .database import get_db
|
||
from .models import Role, User
|
||
from .services import sessions
|
||
|
||
_credentials_exc = HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Nicht angemeldet oder Sitzung abgelaufen",
|
||
)
|
||
|
||
|
||
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
||
token = request.cookies.get(get_settings().cookie_name, "")
|
||
user = sessions.validate(db, token)
|
||
if user is None:
|
||
raise _credentials_exc
|
||
return user
|
||
|
||
|
||
def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
||
if current_user.role != Role.admin:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="Nur Administratoren dürfen diese Aktion ausführen",
|
||
)
|
||
return current_user
|
||
|
||
|
||
def get_owned(db: Session, model, obj_id: int, user: User):
|
||
"""Holt eine Zeile und stellt sicher, dass sie dem Nutzer gehört – sonst 404.
|
||
|
||
Bewusst 404 (nicht 403): fremde IDs sollen nicht einmal als existent gelten.
|
||
Zentrale Stelle für die Mandanten-Isolation."""
|
||
obj = db.get(model, obj_id)
|
||
if obj is None or getattr(obj, "owner_id", None) != user.id:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Nicht gefunden")
|
||
return obj
|