rdmsandClaude Opus 5.5
9e856ec655
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
326 lines
12 KiB
Python
326 lines
12 KiB
Python
from datetime import date
|
||
from decimal import Decimal
|
||
from typing import Literal
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.auth import require_auth
|
||
from app.db import get_session
|
||
from app.engine.loans import loan_schedule
|
||
from app.models.tables import (Category, Loan, PlannedItem, RecurringItem,
|
||
ScenarioLoan)
|
||
from app.services.double_count import (CODE as DOPPELZAEHLUNG_CODE, Konflikt,
|
||
KreditDaten, PostenDaten,
|
||
conflicts_for_loan, conflicts_for_recurring,
|
||
kredit_aus, posten_aus)
|
||
from app.services.suggestions import suggest_recurring
|
||
|
||
router = APIRouter(prefix="/api", tags=["planning"],
|
||
dependencies=[Depends(require_auth)])
|
||
|
||
|
||
# ---------------------------------------------------------------- Recurring
|
||
class RecurringIn(BaseModel):
|
||
name: str
|
||
amount: Decimal
|
||
rhythm: Literal["monthly", "quarterly", "yearly"]
|
||
due_day: int = Field(ge=1, le=31)
|
||
start_date: date | None = None
|
||
end_date: date | None = None
|
||
category_id: int | None = None
|
||
|
||
@model_validator(mode="after")
|
||
def _ende_nicht_vor_start(self):
|
||
if (self.start_date is not None and self.end_date is not None
|
||
and self.end_date < self.start_date):
|
||
raise ValueError("Ende darf nicht vor Start liegen")
|
||
return self
|
||
|
||
|
||
class RecurringCreate(RecurringIn):
|
||
# Nur Anfrage: bestätigt eine erkannte Doppelzählung (nicht gespeichert).
|
||
doppelzaehlung_bestaetigt: bool = False
|
||
|
||
|
||
class RecurringOut(RecurringIn):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
id: int
|
||
|
||
|
||
class RecurringPatch(BaseModel):
|
||
name: str | None = None
|
||
amount: Decimal | None = None
|
||
rhythm: Literal["monthly", "quarterly", "yearly"] | None = None
|
||
due_day: int | None = Field(default=None, ge=1, le=31)
|
||
start_date: date | None = None
|
||
end_date: date | None = None
|
||
category_id: int | None = None
|
||
doppelzaehlung_bestaetigt: bool = False
|
||
|
||
|
||
class SuggestionOut(BaseModel):
|
||
name: str
|
||
amount: Decimal
|
||
rhythm: str
|
||
due_day: int
|
||
start_date: date | None = None
|
||
category_id: int | None = None
|
||
hinweis: str = ""
|
||
|
||
|
||
def _check_category(session: Session, category_id: int | None) -> None:
|
||
if category_id is not None and session.get(Category, category_id) is None:
|
||
raise HTTPException(404, "Kategorie nicht gefunden")
|
||
|
||
|
||
_CENT = Decimal("0.01")
|
||
_FLAG = "doppelzaehlung_bestaetigt"
|
||
|
||
|
||
def _doppelzaehlung(konflikte: list[Konflikt]) -> HTTPException:
|
||
"""409 mit strukturierter Erklärung; die GUI zeigt daraus das Warnfenster."""
|
||
return HTTPException(409, detail={
|
||
"code": DOPPELZAEHLUNG_CODE,
|
||
"meldung": ("Der Betrag entspricht vermutlich einer Kreditrate, die bereits "
|
||
"über „Kredite“ eingeplant ist (oder umgekehrt). Beide zusammen "
|
||
"zählen die Rate doppelt – Planung und Szenarien wären zu "
|
||
"pessimistisch. Nur speichern, wenn das wirklich gewollt ist."),
|
||
"konflikte": [{"posten": k.posten_name,
|
||
"posten_betrag": str(k.posten_betrag.quantize(_CENT)),
|
||
"kredit": k.kredit_name,
|
||
"kredit_rate": str(k.kredit_rate.quantize(_CENT))}
|
||
for k in konflikte],
|
||
})
|
||
|
||
|
||
@router.get("/recurring", response_model=list[RecurringOut])
|
||
def list_recurring(session: Session = Depends(get_session)):
|
||
return [RecurringOut.model_validate(r)
|
||
for r in session.execute(select(RecurringItem)).scalars()]
|
||
|
||
|
||
@router.post("/recurring", response_model=RecurringOut, status_code=201)
|
||
def create_recurring(data: RecurringCreate, session: Session = Depends(get_session)):
|
||
_check_category(session, data.category_id)
|
||
fields = data.model_dump(exclude={_FLAG})
|
||
if not data.doppelzaehlung_bestaetigt:
|
||
konflikte = conflicts_for_recurring(session, PostenDaten(
|
||
None, fields["name"], fields["amount"], fields["rhythm"],
|
||
fields["start_date"], fields["end_date"]))
|
||
if konflikte:
|
||
raise _doppelzaehlung(konflikte)
|
||
item = RecurringItem(**fields)
|
||
session.add(item)
|
||
session.commit()
|
||
session.refresh(item)
|
||
return RecurringOut.model_validate(item)
|
||
|
||
|
||
# Muss vor /recurring/{item_id} stehen, sonst faengt die {id}-Route den Pfad ab.
|
||
@router.get("/recurring/suggestions", response_model=list[SuggestionOut])
|
||
def recurring_suggestions(session: Session = Depends(get_session)):
|
||
return suggest_recurring(session)
|
||
|
||
|
||
@router.patch("/recurring/{item_id}", response_model=RecurringOut)
|
||
def patch_recurring(item_id: int, data: RecurringPatch,
|
||
session: Session = Depends(get_session)):
|
||
item = session.get(RecurringItem, item_id)
|
||
if item is None:
|
||
raise HTTPException(404, "Fixposten nicht gefunden")
|
||
vorher = {k.paar for k in conflicts_for_recurring(session, posten_aus(item))}
|
||
fields = data.model_dump(exclude_unset=True, exclude={_FLAG})
|
||
if "category_id" in fields:
|
||
_check_category(session, fields["category_id"])
|
||
for key, value in fields.items():
|
||
setattr(item, key, value)
|
||
if (item.start_date is not None and item.end_date is not None
|
||
and item.end_date < item.start_date):
|
||
session.rollback()
|
||
raise HTTPException(422, "Ende darf nicht vor Start liegen")
|
||
if not data.doppelzaehlung_bestaetigt:
|
||
neu = [k for k in conflicts_for_recurring(session, posten_aus(item))
|
||
if k.paar not in vorher]
|
||
if neu:
|
||
session.rollback()
|
||
raise _doppelzaehlung(neu)
|
||
session.commit()
|
||
session.refresh(item)
|
||
return RecurringOut.model_validate(item)
|
||
|
||
|
||
@router.delete("/recurring/{item_id}", status_code=204)
|
||
def delete_recurring(item_id: int, session: Session = Depends(get_session)):
|
||
item = session.get(RecurringItem, item_id)
|
||
if item is None:
|
||
raise HTTPException(404, "Fixposten nicht gefunden")
|
||
session.delete(item)
|
||
session.commit()
|
||
|
||
|
||
# ------------------------------------------------------------------ Planned
|
||
class PlannedIn(BaseModel):
|
||
name: str
|
||
amount: Decimal
|
||
due: date
|
||
category_id: int | None = None
|
||
|
||
|
||
class PlannedOut(PlannedIn):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
id: int
|
||
|
||
|
||
class PlannedPatch(BaseModel):
|
||
name: str | None = None
|
||
amount: Decimal | None = None
|
||
due: date | None = None
|
||
category_id: int | None = None
|
||
|
||
|
||
@router.get("/planned", response_model=list[PlannedOut])
|
||
def list_planned(session: Session = Depends(get_session)):
|
||
return [PlannedOut.model_validate(p)
|
||
for p in session.execute(select(PlannedItem)).scalars()]
|
||
|
||
|
||
@router.post("/planned", response_model=PlannedOut, status_code=201)
|
||
def create_planned(data: PlannedIn, session: Session = Depends(get_session)):
|
||
_check_category(session, data.category_id)
|
||
item = PlannedItem(**data.model_dump())
|
||
session.add(item)
|
||
session.commit()
|
||
session.refresh(item)
|
||
return PlannedOut.model_validate(item)
|
||
|
||
|
||
@router.patch("/planned/{item_id}", response_model=PlannedOut)
|
||
def patch_planned(item_id: int, data: PlannedPatch,
|
||
session: Session = Depends(get_session)):
|
||
item = session.get(PlannedItem, item_id)
|
||
if item is None:
|
||
raise HTTPException(404, "Einmalposten nicht gefunden")
|
||
fields = data.model_dump(exclude_unset=True)
|
||
if "category_id" in fields:
|
||
_check_category(session, fields["category_id"])
|
||
for key, value in fields.items():
|
||
setattr(item, key, value)
|
||
session.commit()
|
||
session.refresh(item)
|
||
return PlannedOut.model_validate(item)
|
||
|
||
|
||
@router.delete("/planned/{item_id}", status_code=204)
|
||
def delete_planned(item_id: int, session: Session = Depends(get_session)):
|
||
item = session.get(PlannedItem, item_id)
|
||
if item is None:
|
||
raise HTTPException(404, "Einmalposten nicht gefunden")
|
||
session.delete(item)
|
||
session.commit()
|
||
|
||
|
||
# --------------------------------------------------------------------- Loans
|
||
class LoanIn(BaseModel):
|
||
name: str
|
||
principal: Decimal
|
||
annual_rate_pct: Decimal
|
||
term_months: int
|
||
payout_date: date
|
||
repayment_type: Literal["annuity", "bullet"] = "annuity"
|
||
|
||
|
||
class LoanCreate(LoanIn):
|
||
doppelzaehlung_bestaetigt: bool = False
|
||
|
||
|
||
class LoanOut(LoanIn):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
id: int
|
||
|
||
|
||
class LoanPatch(BaseModel):
|
||
name: str | None = None
|
||
principal: Decimal | None = None
|
||
annual_rate_pct: Decimal | None = None
|
||
term_months: int | None = None
|
||
payout_date: date | None = None
|
||
repayment_type: Literal["annuity", "bullet"] | None = None
|
||
doppelzaehlung_bestaetigt: bool = False
|
||
|
||
|
||
class InstallmentOut(BaseModel):
|
||
due: date
|
||
payment: Decimal
|
||
interest: Decimal
|
||
principal: Decimal
|
||
remaining: Decimal
|
||
|
||
|
||
@router.get("/loans", response_model=list[LoanOut])
|
||
def list_loans(session: Session = Depends(get_session)):
|
||
return [LoanOut.model_validate(l)
|
||
for l in session.execute(select(Loan)).scalars()]
|
||
|
||
|
||
@router.post("/loans", response_model=LoanOut, status_code=201)
|
||
def create_loan(data: LoanCreate, session: Session = Depends(get_session)):
|
||
fields = data.model_dump(exclude={_FLAG})
|
||
if not data.doppelzaehlung_bestaetigt:
|
||
konflikte = conflicts_for_loan(session, KreditDaten(
|
||
None, fields["name"], fields["principal"], fields["annual_rate_pct"],
|
||
fields["term_months"], fields["payout_date"], fields["repayment_type"]))
|
||
if konflikte:
|
||
raise _doppelzaehlung(konflikte)
|
||
loan = Loan(**fields)
|
||
session.add(loan)
|
||
session.commit()
|
||
session.refresh(loan)
|
||
return LoanOut.model_validate(loan)
|
||
|
||
|
||
@router.patch("/loans/{loan_id}", response_model=LoanOut)
|
||
def patch_loan(loan_id: int, data: LoanPatch, session: Session = Depends(get_session)):
|
||
loan = session.get(Loan, loan_id)
|
||
if loan is None:
|
||
raise HTTPException(404, "Kredit nicht gefunden")
|
||
vorher = {k.paar for k in conflicts_for_loan(session, kredit_aus(loan))}
|
||
for key, value in data.model_dump(exclude_unset=True, exclude={_FLAG}).items():
|
||
setattr(loan, key, value)
|
||
if not data.doppelzaehlung_bestaetigt:
|
||
neu = [k for k in conflicts_for_loan(session, kredit_aus(loan))
|
||
if k.paar not in vorher]
|
||
if neu:
|
||
session.rollback()
|
||
raise _doppelzaehlung(neu)
|
||
session.commit()
|
||
session.refresh(loan)
|
||
return LoanOut.model_validate(loan)
|
||
|
||
|
||
@router.delete("/loans/{loan_id}", status_code=204)
|
||
def delete_loan(loan_id: int, session: Session = Depends(get_session)):
|
||
loan = session.get(Loan, loan_id)
|
||
if loan is None:
|
||
raise HTTPException(404, "Kredit nicht gefunden")
|
||
# Remove scenario assignments first: no DB-level cascade exists, so a
|
||
# dangling ScenarioLoan row would raise a foreign-key violation (500 on
|
||
# Postgres).
|
||
session.execute(delete(ScenarioLoan).where(ScenarioLoan.loan_id == loan_id))
|
||
session.delete(loan)
|
||
session.commit()
|
||
|
||
|
||
@router.get("/loans/{loan_id}/schedule", response_model=list[InstallmentOut])
|
||
def get_loan_schedule(loan_id: int, session: Session = Depends(get_session)):
|
||
loan = session.get(Loan, loan_id)
|
||
if loan is None:
|
||
raise HTTPException(404, "Kredit nicht gefunden")
|
||
plan = loan_schedule(Decimal(loan.principal), Decimal(loan.annual_rate_pct),
|
||
loan.term_months, loan.payout_date, loan.repayment_type)
|
||
return [InstallmentOut(due=i.due, payment=i.payment, interest=i.interest,
|
||
principal=i.principal, remaining=i.remaining)
|
||
for i in plan]
|