moneyfyandClaude Opus 5
b586d27b77
- Anmeldung über Argon2id und JWT in httpOnly-Cookies, Refresh mit echter Rotation über die neue Tabelle refresh_token - AuthProvider-Protokoll als Vorbereitung für OIDC, Administrator-Anlage beim Erststart mit erzwungenem Passwortwechsel - CRUD für Konten, Kategorien (zweistufiger Baum), Firmen, Recurrences, Preisversionen, Buchungen, Budgets, Vorlagen und Sparziele - Fälligkeiten mit Overlay-Logik: abrufen, bestätigen, auslassen, zurücksetzen - Kontosalden zum Stichtag, Monatsübersicht mit Plan-Ist-Vergleich - SECRET_KEY jetzt mindestens 32 Zeichen; Platzhalter in Produktion abgelehnt - 61 neue Integrationstests, insgesamt 148 grün Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
"""Kleine Helfer, die sich über alle CRUD-Endpunkte wiederholen."""
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import Select, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.errors import NotFoundError
|
|
from app.db.base import Base
|
|
|
|
# Sprechende Bezeichnungen für Fehlermeldungen in der Oberfläche.
|
|
LABELS: dict[str, str] = {
|
|
"account": "Das Konto",
|
|
"amount_version": "Die Preisversion",
|
|
"budget": "Das Budget",
|
|
"budget_template": "Die Budgetvorlage",
|
|
"category": "Die Kategorie",
|
|
"logo_asset": "Das Logo",
|
|
"merchant": "Die Firma",
|
|
"notification_rule": "Die Benachrichtigungsregel",
|
|
"occurrence": "Die Fälligkeit",
|
|
"recurrence": "Der wiederkehrende Posten",
|
|
"savings_goal": "Das Sparziel",
|
|
"transaction": "Die Buchung",
|
|
}
|
|
|
|
|
|
def label_for(model: type[Base]) -> str:
|
|
return LABELS.get(model.__tablename__, "Der Datensatz")
|
|
|
|
|
|
async def get_or_404[ModelT: Base](
|
|
session: AsyncSession, model: type[ModelT], object_id: int
|
|
) -> ModelT:
|
|
"""Lädt einen Datensatz oder wirft einen 404 mit deutscher Meldung."""
|
|
instance = await session.get(model, object_id)
|
|
if instance is None:
|
|
raise NotFoundError(f"{label_for(model)} mit der ID {object_id} existiert nicht.")
|
|
return instance
|
|
|
|
|
|
def apply_updates(instance: Base, payload: BaseModel) -> Base:
|
|
"""Überträgt nur die tatsächlich gesetzten Felder eines PATCH-Schemas."""
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(instance, field, value)
|
|
return instance
|
|
|
|
|
|
async def count_of(session: AsyncSession, statement: Select[Any]) -> int:
|
|
"""Zählt die Treffer einer Abfrage ohne Sortierung und Seitenbegrenzung."""
|
|
subquery = statement.order_by(None).options().subquery()
|
|
return (await session.execute(select(func.count()).select_from(subquery))).scalar_one()
|