118 lines
3.0 KiB
Python
118 lines
3.0 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.models import B4wExport, Category, ComdirectGiro
|
|
from src.models_db import B4wDb, BaseModelDb, CategoryDb, ComdirectGiroDb
|
|
|
|
|
|
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 = {"hash_id": data.hash_id}
|
|
|
|
parameters_new = {
|
|
"id_b4w": data.id,
|
|
"hash_id": data.hash_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,
|
|
"text": 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,
|
|
)
|
|
|
|
|
|
def upsert_comdirect_giro(
|
|
session: Session, data: ComdirectGiro, do_update=False, do_commit=True
|
|
) -> ComdirectGiroDb | None:
|
|
parameters_existing = {"hash_id": data.hash_id}
|
|
|
|
if not data.date_val:
|
|
return
|
|
|
|
parameters_new = {
|
|
"hash_id": data.hash_id,
|
|
"date_valid": data.date_val,
|
|
"amount": data.amount_eur,
|
|
"date_book": data.date_book,
|
|
"type": data.type,
|
|
"text": data.text,
|
|
}
|
|
|
|
return _upsert(
|
|
session,
|
|
ComdirectGiroDb,
|
|
parameters_existing,
|
|
parameters_new,
|
|
do_update=do_update,
|
|
do_commit=do_commit,
|
|
)
|