Luís Murta
e379d77995
Can now download from all banks registered on the banks/nordigen table and write to the PostgresSQL DB (or any DB, since we're now agnostic). Commented out most of the managers functions until the integration with the new DB client is complete. Set Optional relationships. Remove the DB types as dataclasses, it only increased the verbosity of the types w/o much going for it. Change the name Original to Transaction, since the type is the placeholder for the rest of the transaction information.
137 lines
3.7 KiB
Python
137 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import (
|
|
BigInteger,
|
|
Enum,
|
|
ForeignKey,
|
|
MetaData,
|
|
Numeric,
|
|
String,
|
|
Text,
|
|
)
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
|
|
|
|
from decimal import Decimal
|
|
from typing import Annotated, Optional
|
|
import datetime as dt
|
|
import enum
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
__table_args__ = {"schema": "transactions"}
|
|
metadata = MetaData(
|
|
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 Bank(Base):
|
|
__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(
|
|
back_populates="bank", lazy="joined"
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Bank(name={self.name}, BIC={self.BIC}, type={self.type}, nordigen={self.nordigen})"
|
|
|
|
|
|
bankfk = Annotated[str, mapped_column(Text, ForeignKey(Bank.name))]
|
|
|
|
idpk = Annotated[int, mapped_column(BigInteger, primary_key=True)]
|
|
money = Annotated[Decimal, mapped_column(Numeric(16, 2), nullable=False)]
|
|
|
|
|
|
class Transaction(Base):
|
|
__tablename__ = "originals"
|
|
|
|
id: Mapped[idpk] = mapped_column(autoincrement=True)
|
|
date: Mapped[dt.date]
|
|
description: Mapped[Optional[str]]
|
|
bank: Mapped[bankfk]
|
|
amount: Mapped[money]
|
|
|
|
category: Mapped[Optional[Category]] = relationship(
|
|
back_populates="original", lazy="joined"
|
|
)
|
|
note: Mapped[Optional[Note]] = relationship(back_populates="original")
|
|
tags: Mapped[Optional[set[Tag]]] = relationship(
|
|
back_populates="original", cascade="all, delete-orphan", passive_deletes=True
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Transaction(date={self.date}, description={self.description}, bank={self.bank}, amount={self.amount}, category={self.category})"
|
|
|
|
|
|
idfk = Annotated[
|
|
int, mapped_column(BigInteger, ForeignKey(Transaction.id, ondelete="CASCADE"))
|
|
]
|
|
|
|
|
|
class Category(Base):
|
|
__tablename__ = "categorized"
|
|
|
|
id: Mapped[idfk] = mapped_column(primary_key=True)
|
|
category: Mapped[str]
|
|
|
|
original: Mapped[Transaction] = relationship(back_populates="category")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Category({self.category})"
|
|
|
|
|
|
class Note(Base):
|
|
__tablename__ = "notes"
|
|
|
|
id: Mapped[idfk] = mapped_column(primary_key=True)
|
|
note: Mapped[str]
|
|
|
|
original: Mapped[Transaction] = relationship(back_populates="note")
|
|
|
|
|
|
class Nordigen(Base):
|
|
__tablename__ = "nordigen"
|
|
|
|
name: Mapped[bankfk] = mapped_column(primary_key=True)
|
|
bank_id: Mapped[Optional[str]]
|
|
requisition_id: Mapped[Optional[str]]
|
|
invert: Mapped[Optional[bool]]
|
|
|
|
bank: Mapped[Bank] = relationship(back_populates="nordigen")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"(bank_id={self.bank_id}, requisition_id={self.requisition_id}, invert={self.invert})"
|
|
|
|
|
|
class Tag(Base):
|
|
__tablename__ = "tags"
|
|
|
|
id: Mapped[idfk] = mapped_column(primary_key=True)
|
|
tag: Mapped[str] = mapped_column(primary_key=True)
|
|
|
|
original: Mapped[Transaction] = relationship(back_populates="tags")
|