111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from pydantic import Field, computed_field, field_validator
|
|
from sqlalchemy import (
|
|
Date,
|
|
ForeignKey,
|
|
Integer,
|
|
String,
|
|
)
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from src.constants import IBAN_PATTERN, Currency
|
|
from src.models.category import Category, CategoryDb
|
|
from src.models.common import (
|
|
ExportBaseModel,
|
|
IngestdbBase,
|
|
Sign,
|
|
date2str,
|
|
hashify,
|
|
)
|
|
|
|
__all__ = ["B4wDb", "B4wExport", "Category", "CategoryDb"]
|
|
|
|
|
|
class B4wExport(ExportBaseModel):
|
|
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", gt=0, decimal_places=2)
|
|
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 | None = Field(alias="RmtdAcctIBAN", pattern=IBAN_PATTERN)
|
|
category: Category | None = Field(
|
|
alias="Category",
|
|
examples=[
|
|
"Freizeit",
|
|
"Beruf:Gehalt:super Firma",
|
|
"Lebenshaltung:Lebensmitteleinkäuft",
|
|
"",
|
|
],
|
|
)
|
|
notes: str = Field(alias="Notes")
|
|
|
|
def parameters_to_dict(self) -> dict[str, str]:
|
|
return {
|
|
"Id1": f"{self.id}",
|
|
"OwnrAcctIBAN": f"{self.account_iban!s}",
|
|
"OwnrAcctCcy": f"{self.account_currency!s}",
|
|
"ValDt": date2str(self.date_valid),
|
|
"BookgDt": date2str(self.date_book),
|
|
"Amt": f"{self.amount!s}",
|
|
"AmtCcy": f"{self.amount_currency!s}",
|
|
"CdtDbtInd": f"{self.sign!s}",
|
|
"RmtInf": f"{self.reference}",
|
|
"RmtdNm": f"{self.other_name}",
|
|
"RmtdAcctIBAN": f"{self.other_iban or ''}",
|
|
"Category": f"{self.category.to_str() if self.category else ''}",
|
|
"Notes": f"{self.notes}",
|
|
}
|
|
|
|
@computed_field # type: ignore[prop-decorator]
|
|
@property
|
|
def hash_id(self) -> str:
|
|
return hashify(self.id)
|
|
|
|
@property
|
|
def amount_signed(self) -> Decimal:
|
|
return self.amount * self.sign.to_sign()
|
|
|
|
@field_validator("other_iban", mode="before")
|
|
@classmethod
|
|
def other_iban_blank_is_none(cls, iban_in: str | None) -> str | None:
|
|
if iban_in is None or (isinstance(iban_in, str) and not iban_in.strip()):
|
|
return None
|
|
return iban_in
|
|
|
|
@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 B4wDb(IngestdbBase):
|
|
__tablename__ = "import_b4w"
|
|
|
|
id_b4w: Mapped[int] = mapped_column(Integer, nullable=False, unique=True)
|
|
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
|
|
)
|
|
notes: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
|
|
# relationship
|
|
category = relationship("CategoryDb")
|