Luís Murta
48fae5483d
The `Client` was never closing the `Session` in `DatabaseSession`, which meant that tests that compared transactions and banks would work since attributes loaded only on select would still work. However, this was not the intended behavior. Calling select from the client with using sessions should detach objects completely from a DB session. Therefore, attributes on Transaction, TransactionCategory and Bank are now loaded on a join (lazy parameter).
451 lines
12 KiB
Python
451 lines
12 KiB
Python
from __future__ import annotations
|
|
import datetime as dt
|
|
import decimal
|
|
import enum
|
|
import re
|
|
from typing import Annotated, Any, Callable, 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(init=False, lazy="joined")
|
|
|
|
@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(default=False)
|
|
|
|
category: Mapped[Optional[TransactionCategory]] = relationship(
|
|
back_populates="transaction", default=None, lazy="joined"
|
|
)
|
|
tags: Mapped[set[TransactionTag]] = relationship(default_factory=set, lazy="joined")
|
|
note: Mapped[Optional[Note]] = relationship(
|
|
cascade="all, delete-orphan", passive_deletes=True, default=None, lazy="joined"
|
|
)
|
|
|
|
type: Mapped[str] = mapped_column(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[Optional[bankfk]] = mapped_column(default=None)
|
|
|
|
__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[Optional[idfk]] = mapped_column(default=None)
|
|
|
|
__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 Selector_T(enum.Enum):
|
|
unknown = enum.auto()
|
|
nullifier = enum.auto()
|
|
vacations = enum.auto()
|
|
rules = enum.auto()
|
|
algorithm = enum.auto()
|
|
manual = enum.auto()
|
|
|
|
|
|
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", default=Selector_T.unknown, lazy="joined"
|
|
)
|
|
|
|
transaction: Mapped[Transaction] = relationship(
|
|
back_populates="category", init=False, compare=False
|
|
)
|
|
|
|
@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]] = mapped_column(default=None)
|
|
|
|
@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, unsafe_hash=True):
|
|
__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)
|
|
|
|
|
|
categoryselector = Annotated[
|
|
Selector_T,
|
|
mapped_column(Enum(Selector_T, inherit_schema=True)),
|
|
]
|
|
|
|
|
|
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] = mapped_column(default=Selector_T.unknown)
|
|
|
|
@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, init=False, unsafe_hash=True):
|
|
__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 __init__(self, **kwargs: Any) -> None:
|
|
for k, v in kwargs.items():
|
|
setattr(self, k, v)
|
|
|
|
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: Optional[Any], op: Callable[[Any], bool]) -> 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 __init__(self, name: str, **kwargs: Any) -> None:
|
|
super().__init__(**kwargs)
|
|
self.name = name
|
|
|
|
|
|
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 __init__(self, name: str, **kwargs: Any) -> None:
|
|
super().__init__(**kwargs)
|
|
self.tag = name
|