wlfb/bin
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Szenario-Zeitraum, Neuer Betrag, szenario-eigene Zahlungen – Datenmodell + Migration (A14-T1)

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
rdms
2026-09-30 12:58:06 +02:00
co-authored by Claude Opus 5.5
parent c065ade0d5
commit b0a1ba80a1
4 changed files with 181 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
finance/alembic/versions/e7f1a9c3b214_szenario_zeitraum.py
+54
View File
@@ -0,0 +1,54 @@
"""szenario zeitraum, neuer betrag, szenario-eigene wiederkehrende zahlungen
Revision ID: e7f1a9c3b214
Revises: d5e8b3c7a410
Create Date: 2026-09-30
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'e7f1a9c3b214'
down_revision: Union[str, Sequence[str], None] = 'd5e8b3c7a410'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
FK_PLANNED_CATEGORY = 'fk_scenario_planned_items_category_id'
def upgrade() -> None:
# Bestandsdaten: percent/absolute/remove haben end_date NULL (API hat es
# bisher genullt) und gelten damit unverändert unbegrenzt.
op.add_column('scenario_modifiers', sa.Column('start_date', sa.Date(), nullable=True))
op.add_column('scenario_planned_items',
sa.Column('category_id', sa.Integer(), nullable=True))
op.create_foreign_key(FK_PLANNED_CATEGORY, 'scenario_planned_items',
'categories', ['category_id'], ['id'])
op.create_table(
'scenario_recurring_items',
sa.Column('id', sa.Integer(), primary_key=True),
sa.Column('scenario_id', sa.Integer(), sa.ForeignKey('scenarios.id'), nullable=False),
sa.Column('name', sa.String(length=200), nullable=False),
sa.Column('amount', sa.Numeric(precision=12, scale=2), nullable=False),
sa.Column('rhythm', sa.String(length=20), nullable=False),
sa.Column('due_day', sa.Integer(), nullable=False),
sa.Column('start_date', sa.Date(), nullable=True),
sa.Column('end_date', sa.Date(), nullable=True),
sa.Column('category_id', sa.Integer(), sa.ForeignKey('categories.id'), nullable=True),
)
op.create_index('ix_scenario_recurring_items_scenario_id',
'scenario_recurring_items', ['scenario_id'])
def downgrade() -> None:
# Alte Semantik: "set" gab es nicht, end_date nur bei "ende".
op.execute("DELETE FROM scenario_modifiers WHERE kind = 'set'")
op.execute("UPDATE scenario_modifiers SET end_date = NULL WHERE kind <> 'ende'")
op.drop_index('ix_scenario_recurring_items_scenario_id', 'scenario_recurring_items')
op.drop_table('scenario_recurring_items')
op.drop_constraint(FK_PLANNED_CATEGORY, 'scenario_planned_items', type_='foreignkey')
op.drop_column('scenario_planned_items', 'category_id')
op.drop_column('scenario_modifiers', 'start_date')
finance/app/models/tables.py
+22 -2
View File
@@ -123,9 +123,11 @@ class ScenarioModifier(Base):
target_id: Mapped[int] = mapped_column(Integer)
kind: Mapped[str] = mapped_column(String(20))
value: Mapped[Decimal] = mapped_column(MONEY, default=Decimal("0"))
# Nur fuer kind="ende" gesetzt: Datum, an dem der Ziel-Posten innerhalb
# dieses Szenarios endet (Ausbaustufe 5).
# kind="ende": Datum, an dem der Ziel-Posten in diesem Szenario endet (A5).
# Alle anderen Arten: „Gültig bis" (inklusiv, NULL = unbegrenzt, A14).
end_date: Mapped[date | None] = mapped_column(Date, nullable=True)
# „Gültig ab" (inklusiv, NULL = unbegrenzt); bei kind="ende" immer NULL (A14).
start_date: Mapped[date | None] = mapped_column(Date, nullable=True)
class ScenarioPlannedItem(Base):
@@ -138,6 +140,24 @@ class ScenarioPlannedItem(Base):
name: Mapped[str] = mapped_column(String(200))
amount: Mapped[Decimal] = mapped_column(MONEY)
due: Mapped[date] = mapped_column(Date)
category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)
class ScenarioRecurringItem(Base):
"""Wiederkehrende Zahlung, die NUR in einem bestimmten Szenario zählt -
unabhängig von Scenario.include_recurring. Posten-Einträge
(target_type="recurring") treffen sie nie, Kategorie-Einträge schon.
Ausbaustufe 14."""
__tablename__ = "scenario_recurring_items"
id: Mapped[int] = mapped_column(primary_key=True)
scenario_id: Mapped[int] = mapped_column(ForeignKey("scenarios.id"), index=True)
name: Mapped[str] = mapped_column(String(200))
amount: Mapped[Decimal] = mapped_column(MONEY)
rhythm: Mapped[str] = mapped_column(String(20))
due_day: Mapped[int] = mapped_column(Integer)
start_date: Mapped[date | None] = mapped_column(Date, nullable=True)
end_date: Mapped[date | None] = mapped_column(Date, nullable=True)
category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)
class ProjectionPoint(Base):
finance/tests/test_migration_pg.py
+79
View File
@@ -0,0 +1,79 @@
"""Postgres-only: Alembic-Revision e7f1a9c3b214 (Ausbaustufe 14) hoch/runter
auf einer Wegwerf-DB (Name muss auf '_test' enden). Beispiel:
podman run -d --rm --name a14-pg -e POSTGRES_PASSWORD=wegwerf \
-e POSTGRES_DB=fb_test -p 127.0.0.1:55432:5432 docker.io/library/postgres:17.10
FB_TEST_PG_URL=postgresql+psycopg://postgres:wegwerf@127.0.0.1:55432/fb_test \
.venv/bin/python -m pytest -q tests/test_migration_pg.py
"""
import os
from pathlib import Path
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.engine import make_url
PG_URL = os.environ.get("FB_TEST_PG_URL")
pytestmark = pytest.mark.skipif(not PG_URL, reason="FB_TEST_PG_URL nicht gesetzt")
ROOT = Path(__file__).resolve().parent.parent
@pytest.fixture
def alembic_cfg(monkeypatch):
assert make_url(PG_URL).database.endswith("_test"), "nur Wegwerf-DB *_test!"
monkeypatch.setenv("FB_DATABASE_URL", PG_URL)
from app.config import get_settings
get_settings.cache_clear()
engine = create_engine(PG_URL)
with engine.begin() as conn:
conn.execute(text("DROP SCHEMA public CASCADE; CREATE SCHEMA public"))
cfg = Config(str(ROOT / "alembic.ini"))
cfg.set_main_option("script_location", str(ROOT / "alembic"))
yield cfg, engine
# Schema leeren, damit test_views_pg.py (create_all) nicht auf dem
# herabgestuften Alembic-Stand aufsetzt.
with engine.begin() as conn:
conn.execute(text("DROP SCHEMA public CASCADE; CREATE SCHEMA public"))
engine.dispose()
get_settings.cache_clear()
def test_migration_a14_up_down(alembic_cfg):
cfg, engine = alembic_cfg
command.upgrade(cfg, "d5e8b3c7a410")
with engine.begin() as conn:
conn.execute(text("INSERT INTO scenarios (id, name, description, include_recurring, "
"include_planned) VALUES (1, 'S', '', true, true)"))
conn.execute(text(
"INSERT INTO scenario_modifiers (id, scenario_id, target_type, target_id, kind, "
"value, end_date) VALUES "
"(1, 1, 'recurring', 7, 'ende', 0, '2030-10-15'), "
"(2, 1, 'category', 3, 'percent', 10, NULL)"))
command.upgrade(cfg, "head")
with engine.begin() as conn:
rows = conn.execute(text("SELECT id, kind, value, start_date, end_date "
"FROM scenario_modifiers ORDER BY id")).all()
assert [(r.id, r.kind, str(r.value), r.start_date, str(r.end_date))
for r in rows] == [(1, "ende", "0.00", None, "2030-10-15"),
(2, "percent", "10.00", None, "None")]
conn.execute(text(
"INSERT INTO scenario_modifiers (id, scenario_id, target_type, target_id, kind, "
"value, start_date, end_date) VALUES "
"(3, 1, 'recurring', 7, 'set', -80, '2030-11-01', NULL), "
"(4, 1, 'recurring', 7, 'percent', 50, '2030-11-01', '2030-12-31')"))
conn.execute(text(
"INSERT INTO scenario_recurring_items (scenario_id, name, amount, rhythm, due_day) "
"VALUES (1, 'W', -10, 'monthly', 14)"))
insp = inspect(engine)
assert "scenario_recurring_items" in insp.get_table_names()
assert "category_id" in {c["name"] for c in insp.get_columns("scenario_planned_items")}
command.downgrade(cfg, "d5e8b3c7a410")
with engine.begin() as conn:
rows = conn.execute(text("SELECT id, kind, end_date FROM scenario_modifiers "
"ORDER BY id")).all()
assert [(r.id, r.kind, str(r.end_date)) for r in rows] == [
(1, "ende", "2030-10-15"), (2, "percent", "None"), (4, "percent", "None")]
insp = inspect(engine)
assert "scenario_recurring_items" not in insp.get_table_names()
assert "start_date" not in {c["name"] for c in insp.get_columns("scenario_modifiers")}
finance/tests/test_models.py
+26
View File
@@ -34,3 +34,29 @@ def test_scenario_planned_item_roundtrip(db):
db.commit()
item = db.query(ScenarioPlannedItem).one()
assert item.amount == Decimal("5000.00") and item.scenario_id == sc.id
def test_a14_szenario_felder_roundtrip(db):
from app.models.tables import (Category, Scenario, ScenarioModifier,
ScenarioPlannedItem, ScenarioRecurringItem)
cat = Category(name="A14-Kat")
sc = Scenario(name="A14-Modell", description="")
db.add_all([cat, sc])
db.flush()
db.add_all([
ScenarioModifier(scenario_id=sc.id, target_type="recurring", target_id=1,
kind="set", value=Decimal("-80.00"),
start_date=date(2030, 2, 1), end_date=date(2030, 3, 31)),
ScenarioPlannedItem(scenario_id=sc.id, name="E", amount=Decimal("-5.00"),
due=date(2030, 1, 1), category_id=cat.id),
ScenarioRecurringItem(scenario_id=sc.id, name="W", amount=Decimal("-10.00"),
rhythm="monthly", due_day=14,
start_date=date(2030, 1, 16), end_date=None,
category_id=cat.id),
])
db.commit()
m = db.query(ScenarioModifier).one()
assert (m.kind, m.start_date, m.end_date) == ("set", date(2030, 2, 1), date(2030, 3, 31))
assert db.query(ScenarioPlannedItem).one().category_id == cat.id
w = db.query(ScenarioRecurringItem).one()
assert (w.scenario_id, w.rhythm, w.due_day, w.category_id) == (sc.id, "monthly", 14, cat.id)