db helper for ingesting b4w data

src/helper_db.py

@@ -0,0 +1,77 @@
+#!/bin/python
+
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from src.models import B4wExport, Category
+from src.models_db import B4wDb, BaseModelDb, CategoryDb
+
+
+def _query_existing(
+    session: Session, DbModel: type[BaseModelDb], query_parameters: dict[str, Any]
+) -> BaseModelDb | None:
+    query = session.query(DbModel).filter_by(**query_parameters)
+    return query.first()
+
+
+def _upsert[DbModelT: BaseModelDb](
+    session: Session,
+    DbModel: type[DbModelT],
+    parameters_existing: dict[str, Any],
+    parameters_new: dict[str, Any],
+    do_update: bool = False,
+    do_commit: bool = True,
+) -> DbModelT:
+    existing_db_row = _query_existing(session, DbModel, parameters_existing)
+
+    if existing_db_row:
+        if do_update:
+            for key, value in parameters_new.items():
+                setattr(existing_db_row, key, value)
+        assert isinstance(existing_db_row, DbModel)
+        return existing_db_row
+
+    new_db_row = DbModel(**parameters_new)
+    session.add(new_db_row)
+    session.flush()
+    if do_commit:
+        session.commit()
+    return new_db_row
+
+
+def upsert_category(
+    session: Session, category: Category | None, do_commit=True
+) -> CategoryDb | None:
+    if category is None:
+        return None
+    parameters = {
+        "name": category.name,
+        "parent": upsert_category(session, category.parent),
+    }
+    return _upsert(session, CategoryDb, parameters, parameters, do_update=False, do_commit=do_commit)
+
+
+def upsert_b4w(
+    session: Session, data: B4wExport, do_update=False, do_commit=True
+) -> B4wDb | None:
+    category_db = upsert_category(session, data.category)
+    
+    parameters_existing = {"id_b4w": data.id}
+
+    parameters_new = {
+        "id_b4w": data.id,
+        "account_iban": data.account_iban,
+        "account_currency": data.account_currency.value,
+        "date_valid": data.date_valid,
+        "date_book": data.date_book,
+        "amount": data.amount_signed,
+        "amount_currency": data.amount_currency.value,
+        "reference": data.reference,
+        "other_name": data.other_name,
+        "other_iban": data.other_iban,
+        "category": category_db,
+        "notes": data.notes,
+    }
+    
+    return _upsert(session, B4wDb, parameters_existing, parameters_new, do_update=do_update, do_commit=do_commit)

src/models_db.py

@@ -0,0 +1,57 @@
+#!/bin/python
+
+from sqlalchemy import (
+    Column,
+    Date,
+    ForeignKey,
+    Integer,
+    Numeric,
+    String,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import DeclarativeBase, relationship
+
+
+class BaseModelDb(DeclarativeBase):
+    pass
+
+
+class CategoryDb(BaseModelDb):
+    __tablename__ = "category"
+    __table_args__ = (
+        UniqueConstraint("name", "parent_id", name="uq_category_name_parent"),
+    )
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    name = Column(String, nullable=False)
+    parent_id = Column(Integer, ForeignKey("category.id"), nullable=True)
+
+    # Relationships
+    parent = relationship(
+        "CategoryDb", remote_side="CategoryDb.id", back_populates="children"
+    )
+    children = relationship("CategoryDb", back_populates="parent")
+
+    UniqueConstraint("name", "parent_id", name="uq_category_name_parent")
+
+
+class B4wDb(BaseModelDb):
+    __tablename__ = "import_b4w"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    id_b4w = Column(Integer, nullable=False, unique=True)
+    account_iban = Column(String, nullable=False)
+    account_currency = Column(String(3), nullable=False)
+    date_valid = Column(Date, nullable=False)
+    date_book = Column(Date, nullable=True)
+    amount = Column(Numeric(12, 2), nullable=False)
+    amount_currency = Column(String(3), nullable=False)
+    sign = Column(Integer, nullable=False)
+    reference = Column(String, nullable=True)
+    other_name = Column(String, nullable=True)
+    other_iban = Column(String, nullable=True)
+    category_id = Column(Integer, ForeignKey("category.id"), nullable=True)
+    notes = Column(String, nullable=True)
+
+    # relationship
+    category = relationship("CategoryDb")