Implement exporting transactions to JSONL

import getpass

import json

from pathlib import Path

from typing import Annotated

import typer

from sqlalchemy import select

from poolpay import db, paths

from poolpay.model.BankAccount import BankAccount

from poolpay.model.Transaction import Transaction

app = typer.Typer()

@app.command()

def sync(

    db_path: Annotated[

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

    ] = None,

) -> None:

def sync() -> 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()

@app.command()

def export(

    filter_purpose: Annotated[

        str | None,

        typer.Option(

            help="An optonal regex to keep only transactions with matching purpose"

        ),

    ] = r"Aufladen",

) -> None:

    """

    Export all relevant transactions as messages to be applied in a running PoolPay instance via the Wire server.

    Produces output in JSONL format to stdout.

    """

    base_selection = select(Transaction)

    selection = (

        base_selection.where(Transaction.purpose.regexp_match(filter_purpose))

        if filter_purpose

        else base_selection

    )

    sorted_selection = selection.order_by(Transaction.date)

    transactions = db.session.execute(sorted_selection).scalars().all()

    messages = [

        {

            "uuid": str(transaction.uuid),

            "action": {

                "type": "update_balance",

                "by_amount": transaction.amount_cents,

            },

            "person": {"full_name": transaction.partner},

        }

        for transaction in transactions

    ]

    for message in messages:

        print(json.dumps(message))

if __name__ == "__main__":

    try:

        db.open(_guess_db_path())

        app()

    finally:

        db.close()