Files
certain/backend/app/models.py
T
GuidoandClaude Opus 4.8 1381967aae feat: Phase 4 – CSV-Import (Templates, Prüfansicht, Duplikat/Transfer-Erkennung, Undo)
Backend:
- Modelle + Migration 0005: import_templates, import_batches, categorization_rules;
  transactions um import_batch_id + import_hash erweitert.
- services/csv_import: Datei parsen (Trennzeichen, Dezimal-/Datumsformat, Kodierung,
  Header/Skip, Vorzeichen vs. Soll/Haben), Vorschau mit Duplikat-Erkennung (Hash),
  interner-Transfer-Erkennung via IBAN, lernender Auto-Kategorisierung; Buchen mit
  Batch und Undo.
- routers/imports: Templates (CRUD), Roh-Vorschau, Prüf-Vorschau, Buchen, Verlauf,
  Rückgängig. pytest: Parsing, Duplikate, Transfer, Buchen, Undo, Lernen.

Frontend:
- Import-Seite: Konto+Datei, Format/Spaltenzuordnung (mit Roh-Vorschau),
  editierbare Prüfansicht (Kategorie/Typ/Zielkonto, Duplikate markiert),
  Buchen, Import-Verlauf mit Undo; Vorlagen speichern; i18n DE/EN.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-19 20:06:50 +02:00

489 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""SQLAlchemy-Modelle.
Konventionen (wie in der Vorlage): ``Mapped[...] = mapped_column(...)``, Geld als
Ganzzahl in Cent/Rappen, Enums als kurze VARCHAR-Strings (``native_enum=False``,
ohne DB-CHECK) – das hält Alembic auf SQLite einfach.
Phase 0 umfasst nur Benutzer, Sitzungen und Benutzer-Einstellungen. Fachliche
Tabellen (Konten, Kategorien, Transaktionen, Budget …) kommen in späteren Phasen
hinzu und werden dann alle über ``owner_id`` je Benutzer getrennt.
"""
from __future__ import annotations
import enum
# date/time unter Alias, weil es Spalten namens "date"/"time" gibt, die die
# gleichnamigen Typen sonst in den (späten) Annotationen beschatten.
from datetime import date as date_t, datetime, time as time_t, timezone
from sqlalchemy import (
Boolean,
Date,
DateTime,
Enum as SAEnum,
ForeignKey,
Integer,
Numeric,
String,
Text,
Time,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .database import Base
def _now() -> datetime:
return datetime.now(timezone.utc)
def _enum(enum_cls: type[enum.Enum]):
"""Enum als VARCHAR ohne DB-CHECK – Alembic-freundlich auf SQLite."""
return SAEnum(enum_cls, native_enum=False, create_constraint=False, length=32)
class Role(str, enum.Enum):
admin = "admin"
user = "user"
class AccountType(str, enum.Enum):
girokonto = "girokonto"
sparkonto = "sparkonto"
bargeld = "bargeld"
kreditkarte = "kreditkarte"
depot = "depot"
sonstiges = "sonstiges"
# Kontotypen, deren Ziel-Transfer standardmäßig als budgetierte Ausgabe zählt
# (Sparen/Investment). Andere Transfers sind budgetneutral.
SAVINGS_ACCOUNT_TYPES = {AccountType.sparkonto, AccountType.depot}
class IdentifierKind(str, enum.Enum):
iban = "iban"
account_no = "account_no"
card_no = "card_no"
class CategoryKind(str, enum.Enum):
income = "income"
expense = "expense"
class TxnType(str, enum.Enum):
expense = "expense"
income = "income"
transfer = "transfer"
class BudgetTreatment(str, enum.Enum):
budgeted = "budgeted"
neutral = "neutral"
class AmountMode(str, enum.Enum):
"""Wie der Betrag in der CSV steht."""
signed = "signed" # eine Spalte, mit Vorzeichen (− = Ausgabe)
debit_credit = "debit_credit" # getrennte Soll-/Haben-Spalten
abs_expense = "abs_expense" # Betrag ohne Vorzeichen, immer Ausgabe
class MatchField(str, enum.Enum):
payee = "payee"
purpose = "purpose"
class ImportStatus(str, enum.Enum):
committed = "committed"
undone = "undone"
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
role: Mapped[Role] = mapped_column(_enum(Role), nullable=False, default=Role.user)
display_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
# Vom Admin mit Temporär-Passwort angelegt: beim ersten Login ändern lassen.
must_change_password: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
sessions: Mapped[list[Session]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
settings: Mapped[list[UserSetting]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class Session(Base):
"""Server-seitige Sitzung. Im Cookie liegt nur das Klartext-Token, hier der Hash."""
__tablename__ = "sessions"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
user_agent: Mapped[str | None] = mapped_column(String(255), nullable=True)
user: Mapped[User] = relationship(back_populates="sessions")
class UserSetting(Base):
"""Freie Schlüssel-Wert-Einstellungen je Benutzer.
Bekannte Schlüssel: ``base_currency`` (EUR|CHF), ``budget_start_day`` (int),
``locale`` (de|en), ``theme`` (system|light|dark), ``date_format``.
"""
__tablename__ = "user_settings"
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[str | None] = mapped_column(String(255), nullable=True)
user: Mapped[User] = relationship(back_populates="settings")
# ---------------------------------------------------------------------------
# Phase 1: Konten, Kategorien, Transaktionen, Wechselkurse
# Alle fachlichen Tabellen tragen owner_id und werden nur owner-gescoped gelesen.
# ---------------------------------------------------------------------------
class Account(Base):
__tablename__ = "accounts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
type: Mapped[AccountType] = mapped_column(_enum(AccountType), nullable=False, default=AccountType.girokonto)
currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR")
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="wallet")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#3f51b5")
opening_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
opening_date: Mapped[date_t | None] = mapped_column(Date, nullable=True)
is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
identifiers: Mapped[list[AccountIdentifier]] = relationship(
back_populates="account", cascade="all, delete-orphan"
)
class AccountIdentifier(Base):
"""Erkennungsmerkmal eines Kontos (IBAN/Kontonummer/Kartennummer) für die
Transfer-Erkennung beim Import. ``value_norm`` ist die normalisierte Form."""
__tablename__ = "account_identifiers"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True
)
kind: Mapped[IdentifierKind] = mapped_column(_enum(IdentifierKind), nullable=False, default=IdentifierKind.iban)
value: Mapped[str] = mapped_column(String(64), nullable=False)
value_norm: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
account: Mapped[Account] = relationship(back_populates="identifiers")
__table_args__ = (
UniqueConstraint("account_id", "kind", "value_norm", name="uq_ident_account_kind_value"),
)
class CategoryGroup(Base):
__tablename__ = "category_groups"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
kind: Mapped[CategoryKind] = mapped_column(_enum(CategoryKind), nullable=False, default=CategoryKind.expense)
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="list")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#6b7480")
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
categories: Mapped[list[Category]] = relationship(
back_populates="group", cascade="all, delete-orphan"
)
class Category(Base):
__tablename__ = "categories"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
group_id: Mapped[int] = mapped_column(
ForeignKey("category_groups.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
type: Mapped[CategoryKind] = mapped_column(_enum(CategoryKind), nullable=False, default=CategoryKind.expense)
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="more-horizontal")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#6b7480")
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
group: Mapped[CategoryGroup] = relationship(back_populates="categories")
class Transaction(Base):
__tablename__ = "transactions"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
type: Mapped[TxnType] = mapped_column(_enum(TxnType), nullable=False)
date: Mapped[date_t] = mapped_column(Date, nullable=False, index=True)
time: Mapped[time_t | None] = mapped_column(Time, nullable=True)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True
)
to_account_id: Mapped[int | None] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=True, index=True
)
# Betrag in Kontowährung (bei Transfers: abgebuchter Betrag vom Quellkonto).
amount_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# Nur bei währungsübergreifenden Transfers: gutgeschriebener Betrag am Zielkonto.
to_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# Manuell gesetzter Kurs (Kontowährung je 1 EUR); überschreibt den EZB-Kurs.
fx_rate: Mapped[float | None] = mapped_column(Numeric(18, 8), nullable=True)
# Eingefrorener EUR-Wert von amount_cents zum Buchungsdatum (Pivot für Budget/Dashboard).
amount_eur_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True, index=True
)
# Übersteuert die Standard-Regel für Sparen/Depot-Transfers (budgeted|neutral).
budget_treatment: Mapped[BudgetTreatment | None] = mapped_column(_enum(BudgetTreatment), nullable=True)
# Kategorie, gegen die ein budgetierter Transfer zählt (Sparen/Investment).
budget_category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True
)
payee: Mapped[str | None] = mapped_column(String(255), nullable=True)
description: Mapped[str | None] = mapped_column(String(255), nullable=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
# Import (Phase 4): Herkunft für Undo und Duplikaterkennung.
import_batch_id: Mapped[int | None] = mapped_column(
ForeignKey("import_batches.id", ondelete="SET NULL"), nullable=True, index=True
)
import_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
class ExchangeRate(Base):
"""EZB-Referenzkurs: wie viele Einheiten der Währung ein Euro wert war.
EUR ist implizit 1.0 (keine Zeile). In der Praxis nur CHF. Nicht owner-gescoped
(öffentliche Referenzdaten)."""
__tablename__ = "exchange_rates"
date: Mapped[date_t] = mapped_column(Date, primary_key=True)
currency: Mapped[str] = mapped_column(String(3), primary_key=True)
rate_per_eur: Mapped[float] = mapped_column(Numeric(18, 8), nullable=False)
source: Mapped[str] = mapped_column(String(16), nullable=False, default="ecb")
# ---------------------------------------------------------------------------
# Phase 2: Budget (ein Standard-Budget als Vorlage + Monats-Overrides)
# Planned-Beträge sind in der Hauptwährung des Nutzers.
# ---------------------------------------------------------------------------
class BudgetTemplateLine(Base):
"""Standard-Budget je Kategorie (die editierbare Vorlage)."""
__tablename__ = "budget_template_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
category_id: Mapped[int] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=False
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
__table_args__ = (
UniqueConstraint("owner_id", "category_id", name="uq_tpl_owner_category"),
)
class PlannedIncomeLine(Base):
"""Geplante Einnahme (z. B. Gehalt) mit optionaler Kategorie und Zielkonto."""
__tablename__ = "planned_income_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
label: Mapped[str] = mapped_column(String(120), nullable=False)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True
)
target_account_id: Mapped[int | None] = mapped_column(
ForeignKey("accounts.id", ondelete="SET NULL"), nullable=True
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class BudgetPeriod(Base):
"""Ein Budget-Monat. Existiert erst, sobald der Monat angepasst wurde."""
__tablename__ = "budget_periods"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
label: Mapped[str] = mapped_column(String(7), nullable=False) # YYYY-MM (End-/Hauptmonat)
customized: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
lines: Mapped[list[BudgetPeriodLine]] = relationship(
back_populates="period", cascade="all, delete-orphan"
)
__table_args__ = (
UniqueConstraint("owner_id", "label", name="uq_period_owner_label"),
)
class BudgetPeriodLine(Base):
"""Monatsspezifischer Budget-Wert (Snapshot-on-Customize)."""
__tablename__ = "budget_period_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
budget_period_id: Mapped[int] = mapped_column(
ForeignKey("budget_periods.id", ondelete="CASCADE"), nullable=False, index=True
)
category_id: Mapped[int] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=False
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
period: Mapped[BudgetPeriod] = relationship(back_populates="lines")
__table_args__ = (
UniqueConstraint("budget_period_id", "category_id", name="uq_periodline_period_category"),
)
# ---------------------------------------------------------------------------
# Phase 3: Dashboard-Layout (frei anordbare Karten, JSON, je Benutzer)
# ---------------------------------------------------------------------------
class DashboardLayout(Base):
__tablename__ = "dashboard_layouts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, unique=True, index=True
)
# JSON-Array: [{i, type, x, y, w, h, props}]
layout: Mapped[str] = mapped_column(Text, nullable=False, default="[]")
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
# ---------------------------------------------------------------------------
# Phase 4: CSV-Import (Templates, Batches für Undo, lernende Kategorie-Regeln)
# ---------------------------------------------------------------------------
class ImportTemplate(Base):
__tablename__ = "import_templates"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
delimiter: Mapped[str] = mapped_column(String(2), nullable=False, default=";")
decimal_sep: Mapped[str] = mapped_column(String(1), nullable=False, default=",")
date_format: Mapped[str] = mapped_column(String(32), nullable=False, default="%d.%m.%Y")
encoding: Mapped[str] = mapped_column(String(16), nullable=False, default="utf-8")
has_header: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
skip_rows: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
amount_mode: Mapped[AmountMode] = mapped_column(_enum(AmountMode), nullable=False, default=AmountMode.signed)
# JSON: Feld -> 0-basierter Spaltenindex (oder null). Felder: date, time,
# amount, debit, credit, currency, payee, purpose, counterparty_iban.
column_map: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
class ImportBatch(Base):
__tablename__ = "import_batches"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False
)
template_id: Mapped[int | None] = mapped_column(
ForeignKey("import_templates.id", ondelete="SET NULL"), nullable=True
)
filename: Mapped[str] = mapped_column(String(255), nullable=False, default="")
row_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
imported_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
status: Mapped[ImportStatus] = mapped_column(_enum(ImportStatus), nullable=False, default=ImportStatus.committed)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
class CategorizationRule(Base):
"""Lernende Regel: Empfänger/Zweck enthält Muster -> Kategorie (oder Transfer)."""
__tablename__ = "categorization_rules"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
match_field: Mapped[MatchField] = mapped_column(_enum(MatchField), nullable=False, default=MatchField.payee)
pattern: Mapped[str] = mapped_column(String(255), nullable=False) # normalisiert (lowercase)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=True
)
priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
hits: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
__table_args__ = (
UniqueConstraint("owner_id", "match_field", "pattern", name="uq_rule_owner_field_pattern"),
)