Merge branch 'fints' into 'master'

"""

Connect to a bank account using FinTS

"""

import getpass

from pathlib import Path

from typing import Annotated

import typer

from poolpay import db, paths

from poolpay.model.BankAccount import BankAccount

app = typer.Typer()

def _guess_db_path() -> Path:

    # By default, we use a dedicated path for the database containing FinTS data, as in the common deployment,

    # we run two separate instances of the software for that

    fints_db_path = paths.instance_dir / "poolpay.fints.db"

    # Only if that doesn't exist, we fall back to the generic database location

    return fints_db_path if fints_db_path.exists() else paths.db_file

@app.command()

def sync(

    db_path: Annotated[

        Path | None, typer.Option(help="Path to the database file (guessed by default)")

    ] = None,

) -> None:

    """

    Download transactions for all nw bank accounts and store them in the database.

    """

    try:

        db.open(db_path or _guess_db_path())

        for bank_account in BankAccount.get_all():

            password = getpass.getpass(f"Password for {bank_account.user}: ")

            bank_account.sync(password)

    finally:

        db.close()

if __name__ == "__main__":

    app()

from collections.abc import Generator

from contextlib import contextmanager

from fints.client import FinTS3PinTanClient

from fints.models import SEPAAccount

from schwifty import IBAN

from sqlalchemy.orm import Mapped, mapped_column, relationship

from poolpay import db

from poolpay.model.Base import Base

from poolpay.model.Iban import Iban

from poolpay.model.Transaction import Transaction

class BankAccount(Base):

    """

    A bank account at a German credit institute.

    """

    __tablename__ = "bank_account"

    iban: Mapped[IBAN] = mapped_column(Iban())

    user: Mapped[str]

    # URL for the bank's FinTS endpoint. Not specific to this account, only specific to this bank

    # (and, to be precise, usually even only specific to this brand of bank, as individual bank branches

    # seem to generally not run their own infrastructure). We store it here nevertheless, because we can't be bothered.

    #

    # To find this URL for your bank, you can either search the web (usually, the bank should provide it somewhere),

    # or consult the FinTS Bankenliste (https://www.fints.org/de/hersteller/bankenliste), which is a large Excel file

    # you will receive once you have registered your FinTS product.

    url: Mapped[str]

    # FinTS-Produktregistrierungsnummer for this software. Not specific to this account, only specific to PoolPay itself.

    # We should store it in some general config, but again, couldn't be bothered.

    # Technically, if we hadn't registered the product as "FinTS Kernel" (which seems to be for individual/testing use

    # only), we could have just hardcoded the product ID).

    product_id: Mapped[str]

    transactions: Mapped[list[Transaction]] = relationship(back_populates="account")

    @contextmanager

    def _fints_client(self, password: str) -> Generator[FinTS3PinTanClient]:

        client = FinTS3PinTanClient(

            bank_identifier=self.iban.bank_code,

            user_id=self.user,

            pin=password,

            server=self.url,

            product_id=self.product_id,

        )

        with client:

            yield client

    def _fints_account(self, client: FinTS3PinTanClient) -> SEPAAccount:

        accounts = client.get_sepa_accounts()

        return next(account for account in accounts if account.iban == str(self.iban))

    def sync(self, password: str) -> None:

        """

        Retrieve a list of transactions from the bank, and make sure all of them exist locally.

        """

        with self._fints_client(password) as client:

            account = self._fints_account(client)

            mt940_transactions = client.get_transactions(account)

            for mt940_transaction in mt940_transactions:

                if not Transaction.get_corresponding(mt940_transaction, account=self):

                    transaction = Transaction.from_corresponding(

                        mt940_transaction, account=self

                    )

                    db.session.add(transaction)

        db.session.commit()

import sqlalchemy.types as types

from schwifty import IBAN

from sqlalchemy.engine.interfaces import Dialect

class Iban(types.TypeDecorator[IBAN]):

    impl = types.String

    cache_ok = True

    def process_bind_param(self, value: IBAN | None, dialect: Dialect) -> str | None:

        return str(value) if value is not None else None

    def process_result_value(self, value: str | None, dialect: Dialect) -> IBAN | None:

        return IBAN(value) if value is not None else None

from __future__ import annotations

from datetime import date

from typing import TYPE_CHECKING, Self

from uuid import UUID

import mt940

from sqlalchemy import and_, select

from sqlalchemy.orm import Mapped, mapped_column, relationship

from sqlalchemy.schema import ForeignKey

from poolpay import db

if TYPE_CHECKING:

    from poolpay.model.BankAccount import BankAccount

from poolpay.model.Base import Base

class Transaction(Base):

    """

    A transaction performed at a bank.

    """

    __tablename__ = "transaction"

    date: Mapped[date]

    partner: Mapped[str | None]

    amount_cents: Mapped[int]

    purpose: Mapped[str | None]

    posting_text: Mapped[str]

    account_uuid: Mapped[UUID] = mapped_column(ForeignKey("bank_account.uuid"))

    account: Mapped[BankAccount] = relationship(back_populates="transactions")

    @classmethod

    def get_corresponding(

        cls, transaction: mt940.models.Transaction, account: BankAccount

    ) -> Self | None:

        """

        Returns the Transaction corresponding to the given mt940.models.Transaction,

        or None, if there is no such transaction.

        """

        poolpay_transaction = cls.from_corresponding(transaction, account)

        return db.session.execute(

            select(cls).where(

                and_(

                    cls.date == poolpay_transaction.date,

                    cls.amount_cents == poolpay_transaction.amount_cents,

                    cls.purpose == poolpay_transaction.purpose,

                    cls.partner == poolpay_transaction.partner,

                    cls.posting_text == poolpay_transaction.posting_text,

                )

            )

        ).scalar_one_or_none()

    @classmethod

    def from_corresponding(

        cls, transaction: mt940.models.Transaction, account: BankAccount

    ) -> Self:

        """

        Create a Transaction corresponding to the given mt940.models.Transaction.

        """

        return cls(

            date=transaction.data["date"],

            partner=transaction.data["applicant_name"],

            amount_cents=int(transaction.data["amount"].amount * 100),

            purpose=transaction.data["purpose"],

            posting_text=transaction.data["posting_text"],

            account=account,

        )