248 lines
8.9 KiB
Python
248 lines
8.9 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_validator, model_validator
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import require_auth
|
|
from app.config import get_settings
|
|
from app.db import get_session
|
|
from app.models.tables import (Loan, ProjectionPoint, ProjectionResult,
|
|
Scenario, ScenarioLoan, ScenarioModifier,
|
|
ScenarioPlannedItem)
|
|
from app.services.projection_service import run_projection
|
|
|
|
router = APIRouter(prefix="/api/scenarios", tags=["scenarios"],
|
|
dependencies=[Depends(require_auth)])
|
|
|
|
|
|
class ScenarioIn(BaseModel):
|
|
name: str
|
|
description: str = ""
|
|
include_recurring: bool = True
|
|
include_planned: bool = True
|
|
|
|
|
|
class ScenarioOut(ScenarioIn):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: int
|
|
|
|
|
|
class ScenarioPatch(BaseModel):
|
|
name: str | None = None
|
|
description: str | None = None
|
|
include_recurring: bool | None = None
|
|
include_planned: bool | None = None
|
|
|
|
@field_validator("description")
|
|
@classmethod
|
|
def _leere_beschreibung(cls, v):
|
|
return "" if v is None else v
|
|
|
|
|
|
class ModifierIn(BaseModel):
|
|
target_type: Literal["category", "recurring"]
|
|
target_id: int
|
|
kind: Literal["percent", "absolute", "remove", "ende"]
|
|
value: Decimal = Decimal("0")
|
|
end_date: date | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def _ende_braucht_datum(self):
|
|
if self.kind == "ende" and self.end_date is None:
|
|
raise ValueError("Art »Ende« erfordert ein Datum")
|
|
if self.kind != "ende":
|
|
self.end_date = None
|
|
return self
|
|
|
|
|
|
class ModifierOut(ModifierIn):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: int
|
|
scenario_id: int
|
|
|
|
|
|
class SeriesPointOut(BaseModel):
|
|
day: date
|
|
balance: Decimal
|
|
|
|
|
|
class ProjectionOut(BaseModel):
|
|
low_point_date: date
|
|
low_point_balance: Decimal
|
|
below_zero_date: date | None
|
|
below_threshold_date: date | None
|
|
series: list[SeriesPointOut]
|
|
|
|
|
|
def _get_scenario(session: Session, scenario_id: int) -> Scenario:
|
|
scenario = session.get(Scenario, scenario_id)
|
|
if scenario is None:
|
|
raise HTTPException(404, "Szenario nicht gefunden")
|
|
return scenario
|
|
|
|
|
|
@router.get("", response_model=list[ScenarioOut])
|
|
def list_scenarios(session: Session = Depends(get_session)):
|
|
return [ScenarioOut.model_validate(s)
|
|
for s in session.execute(select(Scenario)).scalars()]
|
|
|
|
|
|
@router.post("", response_model=ScenarioOut, status_code=201)
|
|
def create_scenario(data: ScenarioIn, session: Session = Depends(get_session)):
|
|
if session.execute(select(Scenario).where(Scenario.name == data.name)).scalar():
|
|
raise HTTPException(409, "Szenario existiert bereits")
|
|
scenario = Scenario(**data.model_dump())
|
|
session.add(scenario)
|
|
session.commit()
|
|
session.refresh(scenario)
|
|
return ScenarioOut.model_validate(scenario)
|
|
|
|
|
|
@router.patch("/{scenario_id}", response_model=ScenarioOut)
|
|
def patch_scenario(scenario_id: int, data: ScenarioPatch,
|
|
session: Session = Depends(get_session)):
|
|
scenario = _get_scenario(session, scenario_id)
|
|
fields = data.model_dump(exclude_unset=True)
|
|
if "name" in fields and fields["name"] != scenario.name:
|
|
clash = session.execute(
|
|
select(Scenario).where(Scenario.name == fields["name"])).scalar()
|
|
if clash is not None:
|
|
raise HTTPException(409, "Szenario existiert bereits")
|
|
for key, value in fields.items():
|
|
setattr(scenario, key, value)
|
|
session.commit()
|
|
session.refresh(scenario)
|
|
return ScenarioOut.model_validate(scenario)
|
|
|
|
|
|
@router.delete("/{scenario_id}", status_code=204)
|
|
def delete_scenario(scenario_id: int, session: Session = Depends(get_session)):
|
|
scenario = _get_scenario(session, scenario_id)
|
|
# No DB-level cascades exist, so remove dependent rows explicitly to avoid
|
|
# foreign-key violations (would surface as a 500 on Postgres).
|
|
session.execute(delete(ProjectionPoint).where(
|
|
ProjectionPoint.scenario_id == scenario_id))
|
|
session.execute(delete(ProjectionResult).where(
|
|
ProjectionResult.scenario_id == scenario_id))
|
|
session.execute(delete(ScenarioLoan).where(
|
|
ScenarioLoan.scenario_id == scenario_id))
|
|
session.execute(delete(ScenarioModifier).where(
|
|
ScenarioModifier.scenario_id == scenario_id))
|
|
session.execute(delete(ScenarioPlannedItem).where(
|
|
ScenarioPlannedItem.scenario_id == scenario_id))
|
|
session.delete(scenario)
|
|
session.commit()
|
|
|
|
|
|
@router.post("/{scenario_id}/loans/{loan_id}", status_code=204)
|
|
def add_scenario_loan(scenario_id: int, loan_id: int,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
if session.get(Loan, loan_id) is None:
|
|
raise HTTPException(404, "Kredit nicht gefunden")
|
|
if session.get(ScenarioLoan, (scenario_id, loan_id)) is None:
|
|
session.add(ScenarioLoan(scenario_id=scenario_id, loan_id=loan_id))
|
|
session.commit()
|
|
|
|
|
|
@router.delete("/{scenario_id}/loans/{loan_id}", status_code=204)
|
|
def remove_scenario_loan(scenario_id: int, loan_id: int,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
link = session.get(ScenarioLoan, (scenario_id, loan_id))
|
|
if link is not None:
|
|
session.delete(link)
|
|
session.commit()
|
|
|
|
|
|
@router.post("/{scenario_id}/modifiers", response_model=ModifierOut, status_code=201)
|
|
def add_modifier(scenario_id: int, data: ModifierIn,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
modifier = ScenarioModifier(scenario_id=scenario_id, **data.model_dump())
|
|
session.add(modifier)
|
|
session.commit()
|
|
session.refresh(modifier)
|
|
return ModifierOut.model_validate(modifier)
|
|
|
|
|
|
@router.delete("/{scenario_id}/modifiers/{mod_id}", status_code=204)
|
|
def delete_modifier(scenario_id: int, mod_id: int,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
modifier = session.get(ScenarioModifier, mod_id)
|
|
if modifier is None or modifier.scenario_id != scenario_id:
|
|
raise HTTPException(404, "Modifikator nicht gefunden")
|
|
session.delete(modifier)
|
|
session.commit()
|
|
|
|
|
|
class ScenarioPlannedIn(BaseModel):
|
|
name: str
|
|
amount: Decimal
|
|
due: date
|
|
|
|
|
|
class ScenarioPlannedOut(ScenarioPlannedIn):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: int
|
|
scenario_id: int
|
|
|
|
|
|
@router.get("/{scenario_id}/planned", response_model=list[ScenarioPlannedOut])
|
|
def list_scenario_planned(scenario_id: int, session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
return [ScenarioPlannedOut.model_validate(p) for p in session.execute(
|
|
select(ScenarioPlannedItem)
|
|
.where(ScenarioPlannedItem.scenario_id == scenario_id)
|
|
.order_by(ScenarioPlannedItem.due)).scalars()]
|
|
|
|
|
|
@router.post("/{scenario_id}/planned", response_model=ScenarioPlannedOut,
|
|
status_code=201)
|
|
def add_scenario_planned(scenario_id: int, data: ScenarioPlannedIn,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
item = ScenarioPlannedItem(scenario_id=scenario_id, **data.model_dump())
|
|
session.add(item)
|
|
session.commit()
|
|
session.refresh(item)
|
|
return ScenarioPlannedOut.model_validate(item)
|
|
|
|
|
|
@router.delete("/{scenario_id}/planned/{item_id}", status_code=204)
|
|
def delete_scenario_planned(scenario_id: int, item_id: int,
|
|
session: Session = Depends(get_session)):
|
|
_get_scenario(session, scenario_id)
|
|
item = session.get(ScenarioPlannedItem, item_id)
|
|
if item is None or item.scenario_id != scenario_id:
|
|
raise HTTPException(404, "Einmalzahlung nicht gefunden")
|
|
session.delete(item)
|
|
session.commit()
|
|
|
|
|
|
@router.post("/{scenario_id}/project", response_model=ProjectionOut)
|
|
def project_scenario(scenario_id: int, horizon_days: int | None = None,
|
|
start_date: date | None = None,
|
|
session: Session = Depends(get_session)):
|
|
scenario = _get_scenario(session, scenario_id)
|
|
settings = get_settings()
|
|
resolved_horizon = horizon_days if horizon_days is not None else settings.horizon_days
|
|
resolved_start = start_date if start_date is not None else date.today()
|
|
result = run_projection(session, scenario, resolved_horizon, resolved_start)
|
|
points = session.execute(
|
|
select(ProjectionPoint).where(ProjectionPoint.scenario_id == scenario_id)
|
|
.order_by(ProjectionPoint.day)
|
|
).scalars().all()
|
|
return ProjectionOut(
|
|
low_point_date=result.low_point_date,
|
|
low_point_balance=result.low_point_balance,
|
|
below_zero_date=result.below_zero_date,
|
|
below_threshold_date=result.below_threshold_date,
|
|
series=[SeriesPointOut(day=p.day, balance=p.balance) for p in points],
|
|
)
|