82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from pydantic import Field, computed_field
|
|
from sqlalchemy import Date, ForeignKey, Index, Integer, String, UniqueConstraint
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from src.models.category import Category
|
|
from src.models.common import Amount, ExportBaseModel, IngestdbBase, date2str, hashify
|
|
|
|
|
|
class Transaction(ExportBaseModel):
|
|
"""Canonical, source-agnostic representation of a bank transaction.
|
|
|
|
Both input sources (b4w, comdirect giro) are projected into this model.
|
|
The deduplication key is a business-key hash of the transaction
|
|
attributes, so re-importing an export is idempotent even if
|
|
source-internal ids change between exports.
|
|
"""
|
|
|
|
source: str = Field(description="Source name", examples=["b4w", "comdirect_giro"])
|
|
source_row_id: int = Field(description="Primary key of the staging table row")
|
|
account_iban: str
|
|
account_currency: str
|
|
date_valid: date
|
|
amount: Amount
|
|
amount_currency: str
|
|
date_book: date | None = None
|
|
text: str | None = None
|
|
other_name: str | None = None
|
|
other_iban: str | None = None
|
|
# TODO: The Pydantic model should hold a category object, NOT the category id which is owned by the DB
|
|
category: Category | None
|
|
category_id: int | None = None # remove!
|
|
|
|
@computed_field # type: ignore[prop-decorator]
|
|
@property
|
|
def hash_id(self) -> str:
|
|
return hashify(
|
|
self.source,
|
|
self.account_iban,
|
|
self.date_valid,
|
|
f"{self.amount:.2f}",
|
|
self.text,
|
|
self.other_name,
|
|
)
|
|
|
|
def parameters_to_dict(self) -> dict[str, str]:
|
|
return {
|
|
"source": f"{self.source}",
|
|
"source_row_id": f"{self.source_row_id}",
|
|
"account_iban": f"{self.account_iban}",
|
|
"account_currency": f"{self.account_currency}",
|
|
"date_valid": date2str(self.date_valid),
|
|
"amount": f"{self.amount}",
|
|
"amount_currency": f"{self.amount_currency}",
|
|
"date_book": date2str(self.date_book),
|
|
"text": f"{self.text or ''}",
|
|
"other_name": f"{self.other_name or ''}",
|
|
"other_iban": f"{self.other_iban or ''}",
|
|
"category_id": f"{self.category_id}",
|
|
}
|
|
|
|
|
|
class TransactionDb(IngestdbBase):
|
|
__tablename__ = "transactions"
|
|
__table_args__ = (
|
|
UniqueConstraint("hash_id", name="uq_transactions_hash"),
|
|
Index("ix_transactions_source_row_id", "source", "source_row_id"),
|
|
)
|
|
|
|
source: Mapped[str] = mapped_column(String, nullable=False)
|
|
source_row_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
date_book: Mapped[date | None] = mapped_column(Date, nullable=True)
|
|
other_name: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
other_iban: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
category_id: Mapped[int | None] = mapped_column(
|
|
ForeignKey("category.id"), nullable=True
|
|
)
|
|
|
|
category = relationship("CategoryDb")
|