heavy refactor

src/models/b4w.py

@@ -0,0 +1,120 @@
+from datetime import date
+from decimal import Decimal
+
+from pydantic import BaseModel, Field, computed_field, field_validator
+from sqlalchemy import Column, Date, ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy.orm import relationship
+
+from src.constants import IBAN_PATTERN, Currency
+from src.models._common import BaseModelDb, IngestdbBase, Sign, _hash
+
+
+class B4wExport(BaseModel):
+    id: int = Field(alias="Id1", description="internal B4w ID")
+    account_iban: str = Field(alias="OwnrAcctIBAN", pattern=IBAN_PATTERN)
+    account_currency: Currency = Field(alias="OwnrAcctCcy")
+    date_valid: date = Field(alias="ValDt")
+    date_book: date = Field(alias="BookgDt")
+    amount: Decimal = Field(alias="Amt")
+    amount_currency: Currency = Field(alias="AmtCcy")
+    sign: Sign = Field(alias="CdtDbtInd")
+    reference: str = Field(alias="RmtInf")
+    other_name: str = Field(alias="RmtdNm")
+    other_iban: str = Field(alias="RmtdAcctIBAN")
+    category: Category | None = Field(
+        alias="Category",
+        examples=[
+            "Freizeit",
+            "Beruf:Gehalt:super Firma",
+            "Lebenshaltung:Lebensmitteleinkäuft",
+            "",
+        ],
+    )
+    notes: str = Field(alias="Notes")
+
+    @computed_field
+    @property
+    def hash_id(self) -> str:
+        return _hash(self.id)
+
+    @property
+    def amount_signed(self) -> Decimal:
+        return self.amount * self.sign.to_sign()
+
+    @field_validator("category", mode="before")
+    @classmethod
+    def category_str_is_parsable(
+        cls, category_in: str | Category | None
+    ) -> Category | None:
+        if category_in is None or category_in == "":
+            return None
+        if isinstance(category_in, Category):
+            return category_in
+        category = Category.from_str(category_in)
+        if category is None or category.to_str() != category_in:
+            raise ValueError(f"Category string cannot be parsed: {category_in!r}")
+        return category
+
+
+class Category(BaseModel):
+    name: str = Field(pattern=r"[^:\r\n]+")
+    parent: Category | None = None
+
+    @computed_field
+    @property
+    def hash_id(self) -> str:
+        return _hash(self.name, self.parent)
+
+    @classmethod
+    def from_str(cls, s: str) -> Category | None:
+        if not s:
+            return
+        components = s.split(":")
+        name = components[-1]
+        rest = ":".join(components[:-1])
+        return cls(name=name, parent=cls.from_str(rest))
+
+    def to_str(self) -> str | None:
+        if not self.parent:
+            return self.name
+        else:
+            return f"{self.parent.to_str()}:{self.name}"
+
+    def __str__(self):
+        return self.to_str()
+
+    def __repr__(self):
+        return f'Category.from_str("{self.__str__()}")'
+
+
+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(IngestdbBase):
+    __tablename__ = "import_b4w"
+
+    id_b4w = Column(Integer, nullable=False, unique=True)
+    date_book = Column(Date, 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")