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, Integer, MetaData, Numeric, String, Text, ) from sqlalchemy.orm import ( DeclarativeBase, Mapped, mapped_column, MappedAsDataclass, relationship, ) class Base(MappedAsDataclass, DeclarativeBase): metadata = MetaData( schema="pfbudget", 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): __tablename__ = "banks" name: Mapped[str] = mapped_column(primary_key=True) BIC: Mapped[str] = mapped_column(String(8)) type: Mapped[accounttype] 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().with_variant(Integer, "sqlite"), 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, default=False) type: Mapped[str] = mapped_column(init=False) category: Mapped[Optional[TransactionCategory]] = relationship(init=False) note: Mapped[Optional[Note]] = relationship( cascade="all, delete-orphan", init=False, passive_deletes=True ) tags: Mapped[set[TransactionTag]] = relationship(init=False) __mapper_args__ = {"polymorphic_on": "type", "polymorphic_identity": "transaction"} @property def format(self) -> dict[str, Any]: return dict( id=self.id, 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): __tablename__ = "category_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): __tablename__ = "categories" 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)}," f" 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__ = "transactions_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 if self.selector else None ) class Note(Base): __tablename__ = "notes" id: Mapped[idfk] = mapped_column(primary_key=True, init=False) note: Mapped[str] class Nordigen(Base, Export): __tablename__ = "banks_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): __tablename__ = "tags" 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__ = "transactions_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__ = "category_selectors" 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): __tablename__ = "category_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) start: Mapped[Optional[dt.date]] end: 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, t: BankTransaction) -> bool: valid = None if self.regex: valid = re.compile(self.regex, re.IGNORECASE) ops = ( Rule.exists(self.start, lambda r: r < t.date), Rule.exists(self.end, lambda r: r > t.date), Rule.exists(self.description, lambda r: r == t.description), Rule.exists( valid, lambda r: r.search(t.description) if t.description else False, ), Rule.exists(self.bank, lambda r: r == t.bank), Rule.exists(self.min, lambda r: r < t.amount), Rule.exists(self.max, lambda r: r > t.amount), ) if all(ops): return True return False @property def format(self) -> dict[str, Any]: return dict( start=self.start, end=self.end, description=self.description, regex=self.regex, bank=self.bank, min=self.min, max=self.max, type=self.type, ) @staticmethod def exists(r, op) -> bool: return op(r) if r is not None else True class CategoryRule(Rule): __tablename__ = "category_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): __tablename__ = "tag_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)