central transactions table

src/ingest/transactions.py

@@ -0,0 +1,97 @@
+from collections.abc import Callable
+
+from sqlalchemy import select
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+from sqlalchemy.orm import Session
+
+from src.models._common import BaseModelDb
+from src.models.b4w import B4wDb
+from src.models.comdirect import ComdirectGiroDb
+from src.models.transactions import Transaction, TransactionDb
+
+
+def map_b4w_to_transaction(row: B4wDb) -> Transaction:
+    """Project one staging row of ``import_b4w`` onto a canonical Transaction."""
+    return Transaction(
+        source="b4w",
+        source_row_id=row.id,
+        account_iban=row.account_iban,
+        account_currency=row.account_currency,
+        date_valid=row.date_valid,
+        date_book=row.date_book,
+        amount=row.amount,
+        amount_currency=row.amount_currency,
+        text=row.text,
+        other_name=row.other_name,
+        other_iban=row.other_iban,
+        category_id=row.category_id,
+    )
+
+
+def map_comdirect_giro_to_transaction(row: ComdirectGiroDb) -> Transaction:
+    """Project one staging row of ``comdirect_giro`` onto a canonical Transaction."""
+    return Transaction(
+        source="comdirect_giro",
+        source_row_id=row.id,
+        account_iban=row.account_iban,
+        account_currency=row.account_currency,
+        date_valid=row.date_valid,
+        date_book=row.date_book,
+        amount=row.amount,
+        amount_currency=row.amount_currency,
+        text=row.text,
+    )
+
+
+def upsert_transaction(
+    session: Session, transaction: Transaction, do_commit: bool = True
+) -> None:
+    """Insert one canonical transaction, ignoring duplicates on ``hash_id``.
+
+    Uses a DB-native ``INSERT OR IGNORE`` (SQLite). For PostgreSQL, swap for
+    ``postgresql_insert(TransactionDb).values(...).on_conflict_do_nothing(
+    constraint="uq_transactions_hash")``.
+    """
+    parameters = {
+        "hash_id": transaction.hash_id,
+        "source": transaction.source,
+        "source_row_id": transaction.source_row_id,
+        "account_iban": transaction.account_iban,
+        "account_currency": transaction.account_currency,
+        "date_valid": transaction.date_valid,
+        "date_book": transaction.date_book,
+        "amount": transaction.amount,
+        "amount_currency": transaction.amount_currency,
+        "text": transaction.text,
+        "other_name": transaction.other_name,
+        "other_iban": transaction.other_iban,
+        "category_id": transaction.category_id,
+    }
+    stmt = sqlite_insert(TransactionDb).values(parameters)
+    _ = session.execute(stmt.prefix_with("OR IGNORE"))
+    if do_commit:
+        session.commit()
+
+
+def consolidate_transactions[StagingDbT: BaseModelDb](
+    session: Session,
+    source: str,
+    StagingDb: type[StagingDbT],
+    mapper: Callable[[StagingDbT], Transaction],
+    do_commit: bool = True,
+) -> None:
+    """Project staging rows of one source into the unified ``transactions`` table.
+
+    Idempotent: rows whose business-key hash already exists for this source
+    are skipped, and ``INSERT OR IGNORE`` protects against in-batch duplicates.
+    """
+    existing_hashes = set(
+        session.scalars(select(TransactionDb.hash_id).filter_by(source=source))
+    )
+    for staging_row in session.query(StagingDb).all():
+        transaction = mapper(staging_row)
+        if transaction.hash_id in existing_hashes:
+            continue
+        upsert_transaction(session, transaction, do_commit=False)
+    if do_commit:
+        session.commit()