Transactions table (#7)

Reviewed-on: #7
Co-authored-by: Mathias <mathias2784@web.de>

src/models/transactions.py

@@ -0,0 +1,78 @@
+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._common import ExportBaseModel, IngestdbBase, _hash, date2str
+
+
+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: Decimal = Field(description="Signed amount; negative = money out")
+    amount_currency: str
+    date_book: date | None = None
+    text: str | None = None
+    other_name: str | None = None
+    other_iban: str | None = None
+    category_id: int | None = None
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def hash_id(self) -> str:
+        return _hash(
+            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")