feat: Umsaetze je Konto holen, Takt fuer den automatischen Abruf

Drei Sachen an der Bankanbindung. 'Umsaetze holen' lief ueber alle Konten einer Verbindung. Seit angebundene Konten selbst buchen, ist ein Klick damit folgenreich: Er bucht auf zwei Konten. Der Knopf sitzt jetzt in der Kontozeile; an der Verbindung bleibt 'Alle Konten holen'. Dabei musste der Zeitstempel mitwandern. last_sync_at hing an der Verbindung – ein Einzelabruf haette damit das Startdatum der uebrigen Konten mit vorgeschoben, und deren Umsaetze aus dem uebersprungenen Zeitraum waeren nie angekommen. Der Stempel gehoert an das Konto, das abgerufen wurde. Neu ist ein Takt je Verbindung, nach demselben Muster wie beim Postfach: ein Hintergrund-Thread, der jede Minute nachsieht, ob etwas faellig ist. Bewusst ein eigener Thread – ein haengender Bankzugang darf den Mail-Abruf nicht mitreissen. Die Vorgabe ist 0, also aus: Anders als beim Mail-Abruf entstehen hier Buchungen, das laeuft nicht ungefragt los. Schlaegt ein Abruf fehl, wird der Stempel trotzdem gesetzt, sonst liefe die Verbindung jede Minute in denselben Fehler. Und die Abschnitte heissen jetzt 'Bankkonten Anbindung' und 'Mail Import'. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-27 10:04:29 +02:00
parent a471eba617
commit 8084ba511a
11 changed files with 388 additions and 43 deletions
@@ -106,6 +106,40 @@ async def lifespan(app: FastAPI):
                s.close()

    threading.Thread(target=_mails, name="mail-abruf", daemon=True).start()
+
+    # Bankumsätze im selben Takt-Muster. Bewusst ein EIGENER Thread: Ein
+    # hängender Bankzugang darf den Mail-Abruf nicht mitreissen.
+    #
+    # Anders als beim Mail-Abruf entstehen hier BUCHUNGEN. Deshalb steht der
+    # Takt je Verbindung auf 0, bis ihn jemand einschaltet – nichts läuft
+    # ungefragt los. Gegen doppelte Buchungen sichern die Kennung des Umsatzes,
+    # das Startdatum hinter der letzten Buchung und das Übergehen vorgemerkter
+    # Umsätze; siehe bank_sync.
+    def _bank() -> None:
+        from datetime import datetime, timezone
+
+        from .models import BankConnection, User
+        from .services import bank_sync
+
+        while not stopp.wait(60):
+            s = SessionLocal()
+            try:
+                jetzt = datetime.now(timezone.utc)
+                for verbindung in s.query(BankConnection).filter(
+                    BankConnection.poll_minutes > 0
+                ):
+                    if not bank_sync.faellig(verbindung, jetzt):
+                        continue
+                    nutzer = s.get(User, verbindung.owner_id)
+                    if nutzer is None:
+                        continue
+                    bank_sync.abruf_mit_fehlerbehandlung(s, nutzer, verbindung)
+            except Exception:
+                log.exception("Bank-Abruf fehlgeschlagen")
+            finally:
+                s.close()
+
+    threading.Thread(target=_bank, name="bank-abruf", daemon=True).start()
    yield
    stopp.set()

@@ -881,6 +881,9 @@ class BankConnection(Base):
    # Solange die Autorisierung läuft, steht hier der state-Wert aus /auth.
    pending_state: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    last_sync_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    # Takt des automatischen Abrufs in Minuten, 0 = aus. Wie beim Postfach –
+    # nur ist die Vorgabe hier bewusst aus, denn ein Abruf bucht.
+    poll_minutes: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)

@@ -924,6 +927,12 @@ class BankAccountLink(Base):
    account_id: Mapped[int | None] = mapped_column(
        ForeignKey("accounts.id", ondelete="SET NULL"), nullable=True
    )
+    # Wann DIESES Konto zuletzt abgerufen wurde. An der Verbindung waere der
+    # Stempel falsch, sobald man ein einzelnes Konto abruft: Er schoebe das
+    # Startdatum der uebrigen mit vor, und deren Umsaetze kaemen nie an.
+    last_sync_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)

    __table_args__ = (
@@ -127,6 +127,43 @@ def sync(connection_id: int, db: Session = Depends(get_db),
        raise _fehler(e) from e


+@router.post("/links/{link_id}/sync")
+def sync_link(link_id: int, db: Session = Depends(get_db),
+              user: User = Depends(get_current_user)) -> dict:
+    """Nur die Umsätze DIESES Kontos holen.
+
+    Ein Abruf bucht. Wer ein einzelnes Konto abruft, will nicht, dass nebenbei
+    auf einem zweiten Buchungen entstehen.
+    """
+    verknuepfung = get_owned(db, BankAccountLink, link_id, user)
+    verbindung = get_owned(db, BankConnection, verknuepfung.connection_id, user)
+    try:
+        return bank_sync.hole_umsaetze(db, user, verbindung, nur_link_id=verknuepfung.id)
+    except BankFehler as e:
+        verbindung.last_error = str(e)[:500]
+        db.commit()
+        raise _fehler(e) from e
+
+
+@router.put("/connections/{connection_id}/takt")
+def set_poll(connection_id: int, minutes: int, db: Session = Depends(get_db),
+             user: User = Depends(get_current_user)) -> dict:
+    """Takt des automatischen Abrufs. 0 schaltet ihn ab.
+
+    Untergrenze 15 Minuten: Banken begrenzen unter PSD2 die Zahl der Abrufe je
+    Tag, und häufiger bringt nichts – Umsätze erscheinen nicht im Minutentakt.
+    """
+    verbindung = get_owned(db, BankConnection, connection_id, user)
+    if minutes != 0 and not 15 <= minutes <= 1440:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Der Takt muss 0 (aus) oder zwischen 15 und 1440 Minuten liegen.",
+        )
+    verbindung.poll_minutes = minutes
+    db.commit()
+    return {"poll_minutes": verbindung.poll_minutes}
+
+
@router.post("/connections/{connection_id}/salden")
 def refresh_balances(connection_id: int, db: Session = Depends(get_db),
                     user: User = Depends(get_current_user)) -> dict:
@@ -644,6 +644,8 @@ class BankConnectionOut(BaseModel):
    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

@@ -663,6 +665,7 @@ class BankAccountLinkOut(BaseModel):
    balance_cents: int | None = None
    balance_type: str | None = None
    balance_at: datetime | None = None
+    last_sync_at: datetime | None = None
    account_id: int | None

    model_config = ConfigDict(from_attributes=True)
@@ -363,8 +363,25 @@ def _saldo(salden: list[dict]) -> tuple[int | None, str | None, datetime | None]
    return beste[1], beste[2], beste[3]


+def _verknuepfungen(db: Session, user: User, verbindung: BankConnection,
+                    nur_link_id: int | None) -> list[BankAccountLink]:
+    """Die Konten einer Verbindung – oder genau eines davon.
+
+    Ein Abruf bucht. Wer ein einzelnes Konto abruft, will nicht, dass nebenbei
+    auf einem zweiten Buchungen entstehen.
+    """
+    frage = db.query(BankAccountLink).filter(
+        BankAccountLink.connection_id == verbindung.id,
+        BankAccountLink.owner_id == user.id,
+    )
+    if nur_link_id is not None:
+        frage = frage.filter(BankAccountLink.id == nur_link_id)
+    return frage.all()
+
+
 def hole_salden(db: Session, user: User, verbindung: BankConnection,
-                client: BankClient | None = None) -> int:
+                client: BankClient | None = None,
+                nur_link_id: int | None = None) -> int:
    """Den von der Bank gefuehrten Saldo je Konto auffrischen.

    Auch fuer noch nicht zugeordnete Konten: Bei Revolut-Pockets ist der Betrag
@@ -375,10 +392,7 @@ def hole_salden(db: Session, user: User, verbindung: BankConnection,
    """
    client = client or _client(db, user)
    aktualisiert = 0
-    for verknuepfung in db.query(BankAccountLink).filter(
-        BankAccountLink.connection_id == verbindung.id,
-        BankAccountLink.owner_id == user.id,
-    ):
+    for verknuepfung in _verknuepfungen(db, user, verbindung, nur_link_id):
        try:
            cents, art, stand = _saldo(client.salden(verknuepfung.remote_uid))
        except BankFehler as e:
@@ -416,8 +430,8 @@ def _ist_gebucht(umsatz: dict) -> bool | None:


 def _startdatum(db: Session, user: User, konto: Account,
-                verbindung: BankConnection) -> date:
-    """Ab welchem Tag Umsaetze geholt werden.
+                verknuepfung: BankAccountLink) -> date:
+    """Ab welchem Tag Umsaetze DIESES Kontos geholt werden.

    Beim ERSTEN Abruf ab dem Tag nach der neuesten vorhandenen Buchung des
    Kontos. Ohne das holte die Bank neunzig Tage zurueck – mitten in Monate,
@@ -427,9 +441,13 @@ def _startdatum(db: Session, user: User, konto: Account,

    Spaeter genuegt ein kleiner Rueckgriff hinter den letzten Abruf; einen
    doppelten Abruf faengt die Kennung des Umsatzes exakt ab.
+
+    Der Stempel gehoert an das Konto, nicht an die Verbindung: Sonst schoebe
+    ein Einzelabruf das Startdatum der uebrigen Konten mit vor, und deren
+    Umsaetze aus dem uebersprungenen Zeitraum kaemen nie an.
    """
-    if verbindung.last_sync_at:
-        return verbindung.last_sync_at.date() - timedelta(days=3)
+    if verknuepfung.last_sync_at:
+        return verknuepfung.last_sync_at.date() - timedelta(days=3)
    frueheste = date.today() - timedelta(days=ERSTER_ABRUF_TAGE)
    letzte = (
        db.query(func.max(Transaction.date))
@@ -441,20 +459,20 @@ def _startdatum(db: Session, user: User, konto: Account,
    return max(frueheste, letzte + timedelta(days=1))


-def hole_umsaetze(db: Session, user: User, verbindung: BankConnection) -> dict:
-    """Umsätze aller zugeordneten Konten dieser Verbindung in den Eingang holen."""
+def hole_umsaetze(db: Session, user: User, verbindung: BankConnection,
+                  nur_link_id: int | None = None) -> dict:
+    """Umsätze der zugeordneten Konten dieser Verbindung in den Eingang holen.
+
+    Mit ``nur_link_id`` nur das eine Konto – samt seinem Saldo. Ein Abruf
+    bucht; wer ein Konto abruft, will das andere nicht angefasst haben.
+    """
    client = _client(db, user)
    # Zuerst die Saldi – auch fuer nicht zugeordnete Konten. Sonst saehe man
    # bei drei gleich aussehenden Pockets nie, welches welches ist.
-    saldi = hole_salden(db, user, verbindung, client)
+    saldi = hole_salden(db, user, verbindung, client, nur_link_id)

-    verknuepfungen = [
-        v for v in db.query(BankAccountLink).filter(
-            BankAccountLink.connection_id == verbindung.id,
-            BankAccountLink.owner_id == user.id,
-        )
-        if v.account_id is not None
-    ]
+    verknuepfungen = [v for v in _verknuepfungen(db, user, verbindung, nur_link_id)
+                      if v.account_id is not None]
    if not verknuepfungen:
        return {"geholt": 0, "neu": 0, "saldi": saldi, "ohne_zuordnung": True}

@@ -468,7 +486,7 @@ def hole_umsaetze(db: Session, user: User, verbindung: BankConnection) -> dict:
            continue
        # Je Konto eigen: Das eine ist bis gestern per CSV gepflegt, das andere
        # ganz neu.
-        seit = _startdatum(db, user, konto, verbindung)
+        seit = _startdatum(db, user, konto, verknuepfung)
        ab = seit if ab is None else min(ab, seit)
        for umsatz in client.umsaetze(verknuepfung.remote_uid, seit):
            geholt += 1
@@ -537,6 +555,9 @@ def hole_umsaetze(db: Session, user: User, verbindung: BankConnection) -> dict:
            if eingang.buche(db, user, eintrag) is not None:
                gebucht += 1

+        # Der Stempel gehoert an das Konto, das gerade abgerufen wurde.
+        verknuepfung.last_sync_at = datetime.now(timezone.utc)
+
    verbindung.last_sync_at = datetime.now(timezone.utc)
    verbindung.last_error = None
    db.commit()
@@ -563,6 +584,50 @@ def _datum(umsatz: dict) -> date | None:
    return None


+def _mit_zeitzone(wert: datetime | None) -> datetime | None:
+    """SQLite gibt Zeitstempel ohne Zeitzone zurueck, auch bei ``timezone=True``.
+
+    Ohne diese Ergaenzung scheitert jeder Vergleich mit einem TypeError – und
+    zwar weit weg von der Ursache.
+    """
+    if wert is None or wert.tzinfo is not None:
+        return wert
+    return wert.replace(tzinfo=timezone.utc)
+
+
+def faellig(verbindung: BankConnection, jetzt: datetime) -> bool:
+    """Ist der automatische Abruf dieser Verbindung an der Reihe?
+
+    Bei ``poll_minutes == 0`` nie – nichts laeuft ungefragt los, denn ein Abruf
+    bucht.
+    """
+    if not verbindung.poll_minutes:
+        return False
+    zuletzt = _mit_zeitzone(verbindung.last_sync_at)
+    if zuletzt is None:
+        return True
+    return zuletzt + timedelta(minutes=verbindung.poll_minutes) <= jetzt
+
+
+def abruf_mit_fehlerbehandlung(db: Session, user: User,
+                               verbindung: BankConnection) -> dict:
+    """Wie ``hole_umsaetze``, schreibt einen Fehler aber in die Verbindung.
+
+    Und setzt dabei ``last_sync_at``: Sonst waere die Verbindung sofort wieder
+    faellig und liefe jede Minute in denselben Fehler.
+    """
+    try:
+        return hole_umsaetze(db, user, verbindung)
+    except Exception as e:                 # BankFehler und alles Unerwartete
+        log.warning("Automatischer Abruf fuer %s fehlgeschlagen: %s",
+                    verbindung.aspsp_name, e)
+        db.rollback()
+        verbindung.last_error = str(e)[:500]
+        verbindung.last_sync_at = datetime.now(timezone.utc)
+        db.commit()
+        return {"geholt": 0, "neu": 0, "gebucht": 0, "fehler": str(e)}
+
+
 def laeuft_bald_ab(verbindung: BankConnection) -> bool:
    """Läuft die Zustimmung demnächst ab?

@@ -0,0 +1,37 @@
+"""Abruf je Konto und ein Takt fuer den automatischen Abruf.
+
+``last_sync_at`` hing bisher nur an der Verbindung. Solange immer alle Konten
+zusammen abgerufen wurden, ging das; sobald man ein einzelnes abruft, nicht
+mehr: Der Zeitstempel schoebe das Startdatum der uebrigen Konten mit vor, und
+deren Umsaetze aus dem uebersprungenen Zeitraum kaemen nie an. Der Stempel
+gehoert deshalb an das Konto, das tatsaechlich abgerufen wurde.
+
+``poll_minutes`` ist der Takt des automatischen Abrufs, wie ihn ein Postfach
+schon hat. Vorgabe 0 – nichts laeuft ungefragt los, denn ein Abruf bucht.
+
+Revision ID: 0024
+Revises: 0023
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0024"
+down_revision = "0023"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("bank_account_links") as b:
+        b.add_column(sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True))
+    with op.batch_alter_table("bank_connections") as b:
+        b.add_column(sa.Column("poll_minutes", sa.Integer(), nullable=False,
+                               server_default="0"))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("bank_connections") as b:
+        b.drop_column("poll_minutes")
+    with op.batch_alter_table("bank_account_links") as b:
+        b.drop_column("last_sync_at")
@@ -13,7 +13,7 @@ die Anbindung sonst als Erstes, und der Fehler wäre von aussen nicht zu sehen.
 import base64
 import json
 import secrets
-from datetime import date
+from datetime import date, timedelta

 import pytest
 from cryptography.hazmat.primitives import hashes, serialization
@@ -740,3 +740,98 @@ def test_saldo_kommt_auch_ohne_zuordnung(admin, fake_bank):
    assert r.json()["saldi"] == 1
    link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0]
    assert link["balance_cents"] == 248000
+
+
+def test_abruf_je_konto_ruehrt_das_andere_nicht_an(admin, fake_bank):
+    """Ein Abruf bucht. Wer ein Konto abruft, will nicht, dass nebenbei auf
+    einem zweiten Buchungen entstehen."""
+    _app(admin)
+    eins, zwei = f"acc-{secrets.token_hex(4)}", f"acc-{secrets.token_hex(4)}"
+    heute = date.today().isoformat()
+    fake_bank(
+        konten=[{"uid": eins, "name": "Erstes", "currency": "EUR"},
+                {"uid": zwei, "name": "Zweites", "currency": "EUR"}],
+        umsaetze=[dict(UMSATZ, booking_date=heute)],
+    )
+    state = admin.post("/api/bank/connect",
+                       json={"aspsp_name": f"Zweikonto-{secrets.token_hex(3)}",
+                             "country": "DE"}).json()["state"]
+    admin.post(f"/api/bank/callback?state={state}&code=guter-code")
+
+    links = {v["remote_uid"]: v for v in admin.get("/api/bank/links").json()}
+    ka = _konto(admin, name=f"Konto A {secrets.token_hex(3)}", currency="EUR")
+    kb = _konto(admin, name=f"Konto B {secrets.token_hex(3)}", currency="EUR")
+    admin.post(f"/api/bank/links/{links[eins]['id']}/account/{ka['id']}")
+    admin.post(f"/api/bank/links/{links[zwei]['id']}/account/{kb['id']}")
+
+    r = admin.post(f"/api/bank/links/{links[eins]['id']}/sync")
+    assert r.status_code == 200, r.text
+    assert r.json()["gebucht"] == 1
+
+    assert admin.get(f"/api/accounts/{ka['id']}").json()["balance_cents"] == -1490
+    assert admin.get(f"/api/accounts/{kb['id']}").json()["balance_cents"] == 0, \
+        "das zweite Konto wurde angefasst"
+
+
+def test_einzelabruf_verschiebt_nicht_das_startdatum_des_anderen(admin, fake_bank):
+    """Haenge der Stempel an der Verbindung, schoebe ein Einzelabruf das
+    Startdatum der uebrigen Konten mit vor – deren Umsaetze kaemen nie an."""
+    _app(admin)
+    eins, zwei = f"acc-{secrets.token_hex(4)}", f"acc-{secrets.token_hex(4)}"
+    alt = (date.today() - timedelta(days=30)).isoformat()
+    gegenstelle = fake_bank(
+        konten=[{"uid": eins, "currency": "EUR"}, {"uid": zwei, "currency": "EUR"}],
+        umsaetze=[dict(UMSATZ, booking_date=alt)],
+    )
+    state = admin.post("/api/bank/connect",
+                       json={"aspsp_name": f"Stempel-{secrets.token_hex(3)}",
+                             "country": "DE"}).json()["state"]
+    admin.post(f"/api/bank/callback?state={state}&code=guter-code")
+    links = {v["remote_uid"]: v for v in admin.get("/api/bank/links").json()}
+    ka = _konto(admin, name=f"Stempel A {secrets.token_hex(3)}", currency="EUR")
+    kb = _konto(admin, name=f"Stempel B {secrets.token_hex(3)}", currency="EUR")
+    admin.post(f"/api/bank/links/{links[eins]['id']}/account/{ka['id']}")
+    admin.post(f"/api/bank/links/{links[zwei]['id']}/account/{kb['id']}")
+
+    admin.post(f"/api/bank/links/{links[eins]['id']}/sync")
+    # Das zweite Konto war noch nie dran – es muss weiterhin weit zurueckgreifen.
+    admin.post(f"/api/bank/links/{links[zwei]['id']}/sync")
+    assert gegenstelle.date_from < alt, f"greift nur bis {gegenstelle.date_from} zurueck"
+    assert admin.get(f"/api/accounts/{kb['id']}").json()["balance_cents"] == -1490
+
+
+def test_takt_null_laeuft_nie_und_sonst_erst_wenn_faellig(admin):
+    """Nichts laeuft ungefragt los – ein automatischer Abruf bucht."""
+    from datetime import datetime, timezone
+
+    from app.models import BankConnection
+    from app.services import bank_sync
+
+    jetzt = datetime.now(timezone.utc)
+    aus = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE", poll_minutes=0)
+    assert bank_sync.faellig(aus, jetzt) is False
+
+    nie_gelaufen = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE",
+                                  poll_minutes=60)
+    assert bank_sync.faellig(nie_gelaufen, jetzt) is True
+
+    frisch = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE",
+                            poll_minutes=60,
+                            last_sync_at=jetzt - timedelta(minutes=30))
+    assert bank_sync.faellig(frisch, jetzt) is False
+
+    # SQLite gibt Zeitstempel OHNE Zeitzone zurueck – das darf nicht werfen.
+    ohne_zone = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE",
+                               poll_minutes=60,
+                               last_sync_at=(jetzt - timedelta(hours=2)).replace(tzinfo=None))
+    assert bank_sync.faellig(ohne_zone, jetzt) is True
+
+
+def test_takt_wird_geprueft(admin, fake_bank):
+    verbindung, _, _ = _verbunden(admin, fake_bank)
+    assert admin.put(f"/api/bank/connections/{verbindung}/takt",
+                     params={"minutes": 5}).status_code == 400
+    assert admin.put(f"/api/bank/connections/{verbindung}/takt",
+                     params={"minutes": 0}).json()["poll_minutes"] == 0
+    assert admin.put(f"/api/bank/connections/{verbindung}/takt",
+                     params={"minutes": 240}).json()["poll_minutes"] == 240