from __future__ import annotations import datetime as dt import decimal import enum import re from typing import Annotated, Any, Optional from sqlalchemy import ( BigInteger, Enum, ForeignKey, MetaData, Numeric, String, Text, ) from sqlalchemy.orm import ( DeclarativeBase, Mapped, mapped_column, MappedAsDataclass, relationship, ) class Base(MappedAsDataclass, DeclarativeBase): metadata = MetaData( schema="transactions", naming_convention={ "ix": "ix_%(column_0_label)s", "uq": "uq_%(table_name)s_%(column_0_name)s", "ck": "ck_%(table_name)s_`%(constraint_name)s`", "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", "pk": "pk_%(table_name)s", }, ) class AccountType(enum.Enum): checking = enum.auto() savings = enum.auto() investment = enum.auto() mealcard = enum.auto() VISA = enum.auto() MASTERCARD = enum.auto() accounttype = Annotated[ AccountType, mapped_column(Enum(AccountType, inherit_schema=True)), ] class Export: @property def format(self) -> dict[str, Any]: raise NotImplementedError class Bank(Base, Export): __table_args__ = {"schema": "bank"} __tablename__ = "banks" name: Mapped[str] = mapped_column(unique=True) BIC: Mapped[str] = mapped_column(String(8), primary_key=True) type: Mapped[accounttype] = mapped_column(primary_key=True) nordigen: Mapped[Optional[Nordigen]] = relationship(lazy="joined", init=False) @property def format(self) -> dict[str, Any]: return dict( name=self.name, BIC=self.BIC, type=self.type, nordigen=self.nordigen.format if self.nordigen else None, ) bankfk = Annotated[str, mapped_column(Text, ForeignKey(Bank.name))] idpk = Annotated[int, mapped_column(BigInteger, primary_key=True, autoincrement=True)] money = Annotated[decimal.Decimal, mapped_column(Numeric(16, 2))] class Transaction(Base, Export): __tablename__ = "transactions" id: Mapped[idpk] = mapped_column(init=False) date: Mapped[dt.date] description: Mapped[Optional[str]] amount: Mapped[money] split: Mapped[bool] = mapped_column(init=False) type: Mapped[str] = mapped_column(init=False) category: Mapped[Optional[TransactionCategory]] = relationship(init=False) note: Mapped[Optional[Note]] = relationship(init=False) tags: Mapped[set[TransactionTag]] = relationship(init=False) __mapper_args__ = {"polymorphic_on": "type", "polymorphic_identity": "transaction"} @property def format(self) -> dict[str, Any]: return dict( date=self.date, description=self.description, amount=self.amount, split=self.split, type=self.type, category=self.category.format if self.category else None, # TODO note tags=[tag.format for tag in self.tags] if self.tags else None, ) def __lt__(self, other: Transaction): return self.date < other.date idfk = Annotated[ int, mapped_column(BigInteger, ForeignKey(Transaction.id, ondelete="CASCADE")) ] class BankTransaction(Transaction): bank: Mapped[bankfk] = mapped_column(nullable=True) __mapper_args__ = {"polymorphic_identity": "bank", "polymorphic_load": "inline"} @property def format(self) -> dict[str, Any]: return super().format | dict(bank=self.bank) class MoneyTransaction(Transaction): __mapper_args__ = {"polymorphic_identity": "money"} class SplitTransaction(Transaction): original: Mapped[idfk] = mapped_column(nullable=True) __mapper_args__ = {"polymorphic_identity": "split", "polymorphic_load": "inline"} @property def format(self) -> dict[str, Any]: return super().format | dict(original=self.original) class CategoryGroup(Base, Export): __table_args__ = {"schema": "category"} __tablename__ = "groups" name: Mapped[str] = mapped_column(primary_key=True) @property def format(self) -> dict[str, Any]: return dict(name=self.name) class Category(Base, Export): __table_args__ = {"schema": "category"} __tablename__ = "available" name: Mapped[str] = mapped_column(primary_key=True) group: Mapped[Optional[str]] = mapped_column( ForeignKey(CategoryGroup.name), default=None ) rules: Mapped[set[CategoryRule]] = relationship( cascade="all, delete-orphan", passive_deletes=True, default_factory=set ) schedule: Mapped[Optional[CategorySchedule]] = relationship( cascade="all, delete-orphan", passive_deletes=True, default=None ) def __repr__(self) -> str: return f"Category(name={self.name}, group={self.group}, #rules={len(self.rules)}, schedule={self.schedule})" @property def format(self) -> dict[str, Any]: return dict( name=self.name, group=self.group if self.group else None, rules=[rule.format for rule in self.rules], schedule=self.schedule.format if self.schedule else None, ) catfk = Annotated[ str, mapped_column(ForeignKey(Category.name, ondelete="CASCADE")), ] class TransactionCategory(Base, Export): __tablename__ = "categorized" id: Mapped[idfk] = mapped_column(primary_key=True, init=False) name: Mapped[catfk] selector: Mapped[CategorySelector] = relationship( cascade="all, delete-orphan", lazy="joined" ) @property def format(self): return dict(name=self.name, selector=self.selector.format) class Note(Base): __tablename__ = "notes" id: Mapped[idfk] = mapped_column(primary_key=True, init=False) note: Mapped[str] class Nordigen(Base, Export): __table_args__ = {"schema": "bank"} __tablename__ = "nordigen" name: Mapped[bankfk] = mapped_column(primary_key=True) bank_id: Mapped[Optional[str]] requisition_id: Mapped[Optional[str]] invert: Mapped[Optional[bool]] @property def format(self) -> dict[str, Any]: return dict( name=self.name, bank_id=self.bank_id, requisition_id=self.requisition_id, invert=self.invert, ) class Tag(Base): __table_args__ = {"schema": "tag"} __tablename__ = "available" name: Mapped[str] = mapped_column(primary_key=True) rules: Mapped[set[TagRule]] = relationship( cascade="all, delete-orphan", passive_deletes=True, default_factory=set ) class TransactionTag(Base, Export): __tablename__ = "tagged" id: Mapped[idfk] = mapped_column(primary_key=True, init=False) tag: Mapped[str] = mapped_column(ForeignKey(Tag.name), primary_key=True) @property def format(self): return dict(tag=self.tag) def __hash__(self): return hash(self.id) class Selector_T(enum.Enum): unknown = enum.auto() nullifier = enum.auto() vacations = enum.auto() rules = enum.auto() algorithm = enum.auto() manual = enum.auto() categoryselector = Annotated[ Selector_T, mapped_column(Enum(Selector_T, inherit_schema=True), default=Selector_T.unknown), ] class CategorySelector(Base, Export): __tablename__ = "selector" id: Mapped[int] = mapped_column( BigInteger, ForeignKey(TransactionCategory.id, ondelete="CASCADE"), primary_key=True, init=False, ) selector: Mapped[categoryselector] @property def format(self): return dict(selector=self.selector) class Period(enum.Enum): daily = "daily" weekly = "weekly" monthly = "monthly" yearly = "yearly" scheduleperiod = Annotated[Selector_T, mapped_column(Enum(Period, inherit_schema=True))] class CategorySchedule(Base, Export): __table_args__ = {"schema": "category"} __tablename__ = "schedules" name: Mapped[catfk] = mapped_column(primary_key=True) period: Mapped[Optional[scheduleperiod]] period_multiplier: Mapped[Optional[int]] amount: Mapped[Optional[int]] @property def format(self) -> dict[str, Any]: return dict( name=self.name, period=self.period, period_multiplier=self.period_multiplier, amount=self.amount, ) class Link(Base): __tablename__ = "links" original: Mapped[idfk] = mapped_column(primary_key=True) link: Mapped[idfk] = mapped_column(primary_key=True) class Rule(Base, Export): __tablename__ = "rules" id: Mapped[idpk] = mapped_column(init=False) date: Mapped[Optional[dt.date]] description: Mapped[Optional[str]] regex: Mapped[Optional[str]] bank: Mapped[Optional[str]] min: Mapped[Optional[money]] max: Mapped[Optional[money]] type: Mapped[str] = mapped_column(init=False) __mapper_args__ = { "polymorphic_identity": "rule", "polymorphic_on": "type", } def matches(self, transaction: BankTransaction) -> bool: if ( (self.date and self.date < transaction.date) or ( self.description and transaction.description and self.description not in transaction.description ) or ( self.regex and transaction.description and not re.compile(self.regex, re.IGNORECASE).search( transaction.description ) ) or (self.bank and self.bank != transaction.bank) or (self.min and self.min > transaction.amount) or (self.max and self.max < transaction.amount) ): return False return True @property def format(self) -> dict[str, Any]: return dict( date=self.date, description=self.description, regex=self.regex, bank=self.bank, min=self.min, max=self.max, type=self.type, ) class CategoryRule(Rule): __table_args__ = {"schema": "category"} __tablename__ = "rules" id: Mapped[int] = mapped_column( BigInteger, ForeignKey(Rule.id, ondelete="CASCADE"), primary_key=True, init=False, ) name: Mapped[catfk] __mapper_args__ = { "polymorphic_identity": "category_rule", } @property def format(self) -> dict[str, Any]: return super().format | dict(name=self.name) def __hash__(self): return hash(self.id) class TagRule(Rule): __table_args__ = {"schema": "tag"} __tablename__ = "rules" id: Mapped[int] = mapped_column( BigInteger, ForeignKey(Rule.id, ondelete="CASCADE"), primary_key=True, init=False, ) tag: Mapped[str] = mapped_column(ForeignKey(Tag.name, ondelete="CASCADE")) __mapper_args__ = { "polymorphic_identity": "tag_rule", } @property def format(self) -> dict[str, Any]: return super().format | dict(tag=self.tag) def __hash__(self): return hash(self.id)