"""Pydantic-Schemas für Anfragen/Antworten (Phase 0: Auth, Benutzer, Einstellungen).""" from datetime import date as date_t, datetime, time as time_t from decimal import Decimal from typing import Literal from pydantic import BaseModel, ConfigDict, Field, field_validator from .models import ( AccountType, AmountMode, BudgetTreatment, CategoryKind, IdentifierKind, ImportStatus, Role, TxnType, ) ALLOWED_CURRENCIES = {"EUR", "CHF"} def _check_currency(v: str) -> str: v = v.upper() if v not in ALLOWED_CURRENCIES: raise ValueError("Währung muss EUR oder CHF sein") return v class UserOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int username: str role: Role display_name: str | None = None is_active: bool must_change_password: bool must_change_password_reason: str | None = None totp_enabled: bool = False created_at: datetime # Kein Datenbankfeld: „Richtlinie verlangt MFA und dieser Benutzer hat # keins". Wird beim Ausliefern gesetzt (siehe deps.user_out), damit das # Frontend ohne zweite Abfrage weiss, wohin es umleiten muss. mfa_setup_required: bool = False class UserCreate(BaseModel): username: str = Field(min_length=1, max_length=64) password: str = Field(min_length=1, max_length=1024) role: Role = Role.user display_name: str | None = Field(default=None, max_length=120) class UserUpdate(BaseModel): password: str | None = Field(default=None, min_length=1, max_length=1024) role: Role | None = None display_name: str | None = Field(default=None, max_length=120) is_active: bool | None = None class PasswordChange(BaseModel): current_password: str = Field(min_length=1, max_length=1024) new_password: str = Field(min_length=1, max_length=1024) class SettingOut(BaseModel): key: str value: str | None = None # --- Passwortrichtlinie ----------------------------------------------------- class PolicyOut(BaseModel): min_length: int require_upper: bool require_lower: bool require_digit: bool require_special: bool mfa_required: bool class PolicyIn(BaseModel): min_length: int = Field(ge=8, le=128) require_upper: bool = False require_lower: bool = False require_digit: bool = False require_special: bool = False mfa_required: bool = False # --- Anmeldung / Zwei-Faktor ------------------------------------------------ class LoginOut(BaseModel): """Antwort auf ``POST /auth/login``. ``status`` ist ``"ok"`` (angemeldet, ``user`` gesetzt) oder ``"2fa_required"`` (Passwort stimmt, es fehlt noch der Code – ``user`` bleibt leer).""" status: str = "ok" user: UserOut | None = None class TwoFactorVerify(BaseModel): code: str = Field(min_length=1, max_length=64) recovery: bool = False class TwoFactorSetupOut(BaseModel): secret: str otpauth_uri: str qr_png: str # data:-URI, direkt in verwendbar class TwoFactorActivate(BaseModel): code: str = Field(min_length=1, max_length=16) class TwoFactorPassword(BaseModel): password: str = Field(min_length=1, max_length=1024) class RecoveryCodesOut(BaseModel): codes: list[str] class TwoFactorStatusOut(BaseModel): enabled: bool recovery_codes_left: int # --- Konten ---------------------------------------------------------------- class AccountIdentifierIn(BaseModel): kind: IdentifierKind = IdentifierKind.iban value: str = Field(min_length=1, max_length=64) class AccountIdentifierOut(BaseModel): id: int kind: IdentifierKind value: str class AccountCreate(BaseModel): name: str = Field(min_length=1, max_length=120) type: AccountType = AccountType.girokonto currency: str = "EUR" icon: str = "wallet" color: str = "#3f51b5" opening_balance_cents: int = 0 opening_date: date_t | None = None identifiers: list[AccountIdentifierIn] = Field(default_factory=list) @field_validator("currency") @classmethod def _cur(cls, v: str) -> str: return _check_currency(v) class AccountOrder(BaseModel): """Konto-IDs in der gewünschten Reihenfolge.""" ids: list[int] class AccountUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=120) type: AccountType | None = None currency: str | None = None icon: str | None = None color: str | None = None opening_balance_cents: int | None = None opening_date: date_t | None = None is_archived: bool | None = None position: int | None = None # Wenn gesetzt, ersetzt die komplette Identifikator-Liste. identifiers: list[AccountIdentifierIn] | None = None @field_validator("currency") @classmethod def _cur(cls, v: str | None) -> str | None: return _check_currency(v) if v is not None else None class AccountOut(BaseModel): id: int name: str type: AccountType currency: str icon: str color: str opening_balance_cents: int opening_date: date_t | None is_archived: bool position: int identifiers: list[AccountIdentifierOut] balance_cents: int balance_base_cents: int # Saldo, wie ihn die Bank fuehrt – nur bei angebundenen Konten gesetzt. Er # ersetzt die eigene Rechnung nicht, er prueft sie. bank_balance_cents: int | None = None bank_balance_at: datetime | None = None # Nur ein GEBUCHTER Saldo ist mit Certains Rechnung vergleichbar. Ein # verfuegbarer enthaelt Vorgemerktes, das hier bewusst nicht gebucht wird. bank_balance_vergleichbar: bool = False # --- Kategorien ------------------------------------------------------------ class CategoryOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int group_id: int name: str type: CategoryKind icon: str color: str position: int is_builtin: bool is_archived: bool excluded_from_stats: bool = False # Anzahl hinterlegter Schlagwörter. Nur in der Liste gefüllt, damit die # Übersicht ohne einen Abruf je Kategorie auskommt. keyword_count: int = 0 class CategoryGroupOut(BaseModel): id: int name: str kind: CategoryKind icon: str color: str position: int is_builtin: bool categories: list[CategoryOut] class CategoryGroupCreate(BaseModel): name: str = Field(min_length=1, max_length=120) kind: CategoryKind = CategoryKind.expense icon: str = "list" color: str = "#6b7480" class CategoryGroupUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=120) icon: str | None = None color: str | None = None position: int | None = None class CategoryCreate(BaseModel): group_id: int name: str = Field(min_length=1, max_length=120) icon: str = "more-horizontal" color: str = "#6b7480" excluded_from_stats: bool = False class CategoryUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=120) group_id: int | None = None icon: str | None = None color: str | None = None position: int | None = None is_archived: bool | None = None excluded_from_stats: bool | None = None # --- Transaktionen --------------------------------------------------------- class TransactionCreate(BaseModel): type: TxnType date: date_t time: time_t | None = None account_id: int to_account_id: int | None = None amount_cents: int = Field(gt=0) to_amount_cents: int | None = Field(default=None, gt=0) fx_rate: float | None = Field(default=None, gt=0) category_id: int | None = None budget_treatment: BudgetTreatment | None = None budget_category_id: int | None = None # Budget-Monat (YYYY-MM). Leer heisst: aus dem Datum ableiten. budget_period: str | None = Field(default=None, pattern=r"^\d{4}-\d{2}$") payee: str | None = Field(default=None, max_length=255) description: str | None = Field(default=None, max_length=255) notes: str | None = None class TransactionUpdate(BaseModel): type: TxnType | None = None date: date_t | None = None time: time_t | None = None account_id: int | None = None to_account_id: int | None = None amount_cents: int | None = Field(default=None, gt=0) to_amount_cents: int | None = Field(default=None, gt=0) fx_rate: float | None = Field(default=None, gt=0) category_id: int | None = None budget_treatment: BudgetTreatment | None = None budget_category_id: int | None = None # Budget-Monat (YYYY-MM). Leer heisst: aus dem Datum ableiten. budget_period: str | None = Field(default=None, pattern=r"^\d{4}-\d{2}$") payee: str | None = Field(default=None, max_length=255) description: str | None = Field(default=None, max_length=255) notes: str | None = None class TransactionOut(BaseModel): id: int type: TxnType date: date_t time: time_t | None account_id: int to_account_id: int | None account_currency: str amount_cents: int to_amount_cents: int | None fx_rate: float | None amount_eur_cents: int amount_base_cents: int category_id: int | None budget_treatment: BudgetTreatment | None budget_category_id: int | None # Der Budget-Monat, in dem die Buchung zaehlt – immer gesetzt. Ob er von # Hand gewaehlt wurde oder aus dem Datum stammt, sagt das zweite Feld. budget_period: str budget_period_manuell: bool = False # Woher die Buchung stammt, fuer die Arbeitsliste: Ist sie gesetzt, # kam die Buchung aus einem Datei-Import. import_batch_id: int | None = None # Woher die Buchung stammt: bank, mail, shortcut, transfer, csv, manuell. herkunft: str = "manuell" payee: str | None description: str | None notes: str | None class RateStatus(BaseModel): currency: str count: int min_date: date_t | None max_date: date_t | None latest_rate: float | None class ConversionOut(BaseModel): amount_cents: int from_currency: str to_currency: str date: date_t converted_cents: int | None rate: float | None # --- Budget: geplante Einnahmen -------------------------------------------- class PlannedIncomeCreate(BaseModel): label: str = Field(min_length=1, max_length=120) category_id: int | None = None target_account_id: int | None = None planned_cents: int = Field(ge=0) class PlannedIncomeUpdate(BaseModel): label: str | None = Field(default=None, min_length=1, max_length=120) category_id: int | None = None target_account_id: int | None = None planned_cents: int | None = Field(default=None, ge=0) class PlannedIncomeOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int label: str category_id: int | None target_account_id: int | None planned_cents: int position: int # --- CSV-Import ------------------------------------------------------------ class ImportTemplateCreate(BaseModel): name: str = Field(min_length=1, max_length=120) delimiter: str = ";" decimal_sep: str = "," date_format: str = "%d.%m.%Y" encoding: str = "utf-8" has_header: bool = True skip_rows: int = 0 amount_mode: AmountMode = AmountMode.signed column_map: dict[str, int | None] = Field(default_factory=dict) filter_column: int | None = None filter_value: str | None = None class ImportTemplateUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=120) delimiter: str | None = None decimal_sep: str | None = None date_format: str | None = None encoding: str | None = None has_header: bool | None = None skip_rows: int | None = None amount_mode: AmountMode | None = None column_map: dict[str, int | None] | None = None filter_column: int | None = None filter_value: str | None = None class ImportTemplateOut(BaseModel): id: int name: str delimiter: str decimal_sep: str date_format: str encoding: str has_header: bool skip_rows: int amount_mode: AmountMode column_map: dict[str, int | None] filter_column: int | None = None filter_value: str | None = None class ImportBatchOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int account_id: int template_id: int | None preset: str | None filename: str row_count: int imported_count: int status: ImportStatus created_at: datetime class ImportCommitIn(BaseModel): account_id: int template_id: int | None = None preset: str | None = None filename: str = "" rows: list[dict] = Field(default_factory=list) # --- Schlagwörter (manuelle Kategorisierungs-Regeln) ----------------------- class KeywordCreate(BaseModel): word: str = Field(min_length=1, max_length=255) amount_cents: int | None = Field(default=None, gt=0) class KeywordUpdate(BaseModel): word: str = Field(min_length=1, max_length=255) amount_cents: int | None = Field(default=None, gt=0) class KeywordOut(BaseModel): id: int category_id: int word: str amount_cents: int | None = None class TransferKeywordCreate(BaseModel): word: str = Field(min_length=1, max_length=255) amount_cents: int | None = Field(default=None, gt=0) target_account_id: int | None = None class TransferKeywordOut(BaseModel): id: int word: str amount_cents: int | None = None target_account_id: int | None = None class FlaggedKeywordOut(BaseModel): id: int category_id: int | None = None category_name: str word: str applied_count: int overridden_count: int # --- Gerätemarken und Eingang ------------------------------------------------ class ApiTokenCreate(BaseModel): name: str = Field(min_length=1, max_length=80) default_account_id: int | None = None class ApiTokenOut(BaseModel): id: int name: str default_account_id: int | None last_used_at: datetime | None created_at: datetime | None model_config = ConfigDict(from_attributes=True) class ApiTokenCreated(ApiTokenOut): """Antwort beim Anlegen – der Klartext ist hier das einzige Mal zu sehen.""" token: str class QuickBookingIn(BaseModel): """Was ein Kurzbefehl schickt. Bewusst genügsam: Betrag reicht.""" amount: Decimal payee: str | None = Field(default=None, max_length=255) date: date_t | None = None account_id: int | None = None note: str | None = None # Kennung des Kurzbefehls, damit ein doppelt ausgelöster Aufruf nicht # zweimal bucht (schlechtes Netz, versehentlicher Doppeltipp). external_id: str | None = Field(default=None, max_length=255) kind: Literal["expense", "income"] = "expense" class InboxItemOut(BaseModel): id: int source: str status: str account_id: int | None date: date_t amount_cents: int currency: str payee: str | None note: str | None category_id: int | None transaction_id: int | None # Das jeweils ANDERE Konto, wenn die Buchung ein Kontoübertrag ist. Ohne # das kann die Oberfläche einen Übertrag nicht von einer Ausgabe # unterscheiden und fragt weiter nach einer Kategorie. transfer_account_id: int | None = None # Was die Bank ueber die Umrechnung mitteilt – ihr Kurs schlaegt die EZB. fx_rate: float | None = None counter_amount_cents: int | None = None counter_currency: str | None = None created_at: datetime | None model_config = ConfigDict(from_attributes=True) class QuickBookingOut(BaseModel): """Kurze Rückmeldung – der Kurzbefehl zeigt sie als Mitteilung an.""" ok: bool = True message: str item: InboxItemOut # --- Postfächer für den Mail-Abruf ------------------------------------------- class MailSourceIn(BaseModel): name: str = Field(default="", max_length=80) host: str = Field(min_length=1, max_length=255) port: int = 993 security: Literal["ssl", "starttls", "none"] = "ssl" username: str = Field(min_length=1, max_length=255) # Beim Ändern weglassen, um das gespeicherte Passwort zu behalten. password: str | None = None folder: str = "INBOX" done_folder: str | None = None mark_seen: bool = True is_active: bool = True poll_minutes: int = Field(default=15, ge=5, le=1440) class MailSourceOut(BaseModel): id: int name: str host: str port: int security: str username: str folder: str done_folder: str | None mark_seen: bool is_active: bool poll_minutes: int last_poll_at: datetime | None last_error: str | None # Das Passwort verlässt den Server nie – nur die Tatsache, dass eines da ist. password_set: bool = True model_config = ConfigDict(from_attributes=True) class MailSenderIn(BaseModel): name: str = Field(default="", max_length=80) from_contains: str | None = Field(default=None, max_length=255) subject_contains: str | None = Field(default=None, max_length=255) currency: str | None = Field(default=None, max_length=3) account_id: int | None = None class MailSenderOut(MailSenderIn): id: int source_id: int model_config = ConfigDict(from_attributes=True) # --- Bankanbindung (PSD2) ---------------------------------------------------- class BankAppIn(BaseModel): application_id: str = Field(min_length=1, max_length=64) # Inhalt der .pem-Datei aus dem Control Panel. private_key: str = Field(min_length=1) # Öffentliche Adresse dieser Installation; die Bank leitet dorthin zurück. base_url: str = Field(min_length=1, max_length=255) @field_validator("base_url") @classmethod def nur_https(cls, v: str) -> str: """Ohne HTTPS lehnt die Bank die Rückleitung ab. Bewusst eingetragen statt erkannt: Hinter einem Proxy sieht der Dienst selbst „http", obwohl der Browser HTTPS spricht – automatisches Raten würde hier zuverlässig das Falsche liefern. """ v = v.strip().rstrip("/") if not v.lower().startswith("https://"): raise ValueError( "Die Adresse muss mit https:// beginnen. Banken leiten nicht " "auf eine unverschlüsselte Adresse zurück." ) return v class BankAppOut(BaseModel): id: int application_id: str base_url: str created_at: datetime | None # Der Schlüssel verlässt den Server nie – nur die Tatsache, dass einer da ist. key_set: bool = True model_config = ConfigDict(from_attributes=True) class BankConnectionOut(BaseModel): id: int aspsp_name: str aspsp_country: str valid_until: datetime | None last_sync_at: datetime | None # Takt des automatischen Abrufs in Minuten, 0 = aus. poll_minutes: int = 0 last_error: str | None expires_soon: bool = False model_config = ConfigDict(from_attributes=True) class BankAccountLinkOut(BaseModel): id: int connection_id: int remote_uid: str iban: str | None remote_name: str | None product: str | None = None other_id: str | None = None bank_felder: dict = {} currency: str balance_cents: int | None = None balance_type: str | None = None balance_vergleichbar: bool = False balance_at: datetime | None = None last_sync_at: datetime | None = None account_id: int | None model_config = ConfigDict(from_attributes=True) class BankConnectStart(BaseModel): aspsp_name: str = Field(min_length=1, max_length=120) country: str = Field(min_length=2, max_length=2) psu_type: Literal["personal", "business"] = "personal"