personal-finance/budget
2
1
Fork 0
Code Issues 11 Pull Requests Projects 6 Releases 2 Wiki Activity

Compare commits

base: personal-finance:92f8493f86e3a32f46b74361268a5205d14c7a61
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1
..
compare: personal-finance:fd6793b4f426a52b40fde385243a5ddfe6878ad8
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1

No commits in common. "92f8493f86e3a32f46b74361268a5205d14c7a61" and "fd6793b4f426a52b40fde385243a5ddfe6878ad8" have entirely different histories.
92f8493f86 ... fd6793b4f4

8 changed files with 41 additions and 718 deletions
Show Stats Expand all files Collapse all files

2
alembic/env.py
View File

@ -27,7 +27,7 @@ target_metadata = Base.metadata
def include_name(name, type_, parent_names):
if type_ == "schema":
return name in ["bank", "category", "tag", "transactions"]
return name in ["transactions"]
else:
return True

88
alembic/versions/28556ab17c56_selector_back_to_transaction.py
View File

@ -1,88 +0,0 @@
"""Selector back to transaction
Revision ID: 28556ab17c56
Revises: e455c78df789
Create Date: 2023-01-23 00:34:39.062562+00:00
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = "28556ab17c56"
down_revision = "e455c78df789"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.rename_table("tags", "tagged", schema="transactions")
op.create_table(
"selector",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column(
"selector",
sa.Enum(
"unknown",
"nullifier",
"vacations",
"rules",
"algorithm",
"manual",
name="selector_t",
schema="transactions",
inherit_schema=True,
),
nullable=False,
),
sa.ForeignKeyConstraint(
["id"],
["transactions.categorized.id"],
name=op.f("fk_selector_id_categorized"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_selector")),
schema="transactions",
)
op.drop_table("selector", schema="category")
op.rename_table("originals", "transactions", schema="transactions")
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.rename_table("transactions", "originals", schema="transactions")
op.create_table(
"selector",
sa.Column("id", sa.BIGINT(), autoincrement=False, nullable=False),
sa.Column(
"selector",
postgresql.ENUM(
"unknown",
"nullifier",
"vacations",
"rules",
"algorithm",
"manual",
name="selector_t",
schema="category",
),
autoincrement=False,
nullable=False,
),
sa.ForeignKeyConstraint(
["id"],
["transactions.categorized.id"],
name="fk_selector_id_categorized",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name="pk_selector"),
schema="category",
)
op.drop_table("selector", schema="transactions")
op.rename_table("tagged", "tags", schema="transactions")
# ### end Alembic commands ###

452
alembic/versions/e455c78df789_divide_by_schemas.py
View File

@ -1,452 +0,0 @@
"""Divide by schemas
Revision ID: e455c78df789
Revises: 6b293f78cc97
Create Date: 2023-01-22 23:38:23.266906+00:00
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = "e455c78df789"
down_revision = "6b293f78cc97"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"banks",
sa.Column("name", sa.String(), nullable=False),
sa.Column("BIC", sa.String(length=8), nullable=False),
sa.Column(
"type",
sa.Enum(
"checking",
"savings",
"investment",
"mealcard",
"VISA",
"MASTERCARD",
name="accounttype",
schema="bank",
inherit_schema=True,
),
nullable=False,
),
sa.PrimaryKeyConstraint("BIC", "type", name=op.f("pk_banks")),
sa.UniqueConstraint("name", name=op.f("uq_banks_name")),
schema="bank",
)
op.create_table(
"groups",
sa.Column("name", sa.String(), nullable=False),
sa.PrimaryKeyConstraint("name", name=op.f("pk_groups")),
schema="category",
)
op.create_table(
"available",
sa.Column("name", sa.String(), nullable=False),
sa.PrimaryKeyConstraint("name", name=op.f("pk_available")),
schema="tag",
)
op.create_table(
"nordigen",
sa.Column("name", sa.Text(), nullable=False),
sa.Column("bank_id", sa.String(), nullable=True),
sa.Column("requisition_id", sa.String(), nullable=True),
sa.Column("invert", sa.Boolean(), nullable=True),
sa.ForeignKeyConstraint(
["name"], ["bank.banks.name"], name=op.f("fk_nordigen_name_banks")
),
sa.PrimaryKeyConstraint("name", name=op.f("pk_nordigen")),
schema="bank",
)
op.create_table(
"available",
sa.Column("name", sa.String(), nullable=False),
sa.Column("group", sa.String(), nullable=True),
sa.ForeignKeyConstraint(
["group"], ["category.groups.name"], name=op.f("fk_available_group_groups")
),
sa.PrimaryKeyConstraint("name", name=op.f("pk_available")),
schema="category",
)
op.create_table(
"rules",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column("tag", sa.String(), nullable=False),
sa.ForeignKeyConstraint(
["id"],
["transactions.rules.id"],
name=op.f("fk_rules_id_rules"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["tag"],
["tag.available.name"],
name=op.f("fk_rules_tag_available"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_rules")),
schema="tag",
)
op.create_table(
"rules",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.ForeignKeyConstraint(
["id"],
["transactions.rules.id"],
name=op.f("fk_rules_id_rules"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["name"],
["category.available.name"],
name=op.f("fk_rules_name_available"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_rules")),
schema="category",
)
op.create_table(
"schedules",
sa.Column("name", sa.String(), nullable=False),
sa.Column(
"period",
sa.Enum(
"daily",
"weekly",
"monthly",
"yearly",
name="period",
schema="category",
inherit_schema=True,
),
nullable=True,
),
sa.Column("period_multiplier", sa.Integer(), nullable=True),
sa.Column("amount", sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(
["name"],
["category.available.name"],
name=op.f("fk_schedules_name_available"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("name", name=op.f("pk_schedules")),
schema="category",
)
op.create_table(
"selector",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column(
"selector",
sa.Enum(
"unknown",
"nullifier",
"vacations",
"rules",
"algorithm",
"manual",
name="selector_t",
schema="category",
inherit_schema=True,
),
nullable=False,
),
sa.ForeignKeyConstraint(
["id"],
["transactions.categorized.id"],
name=op.f("fk_selector_id_categorized"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_selector")),
schema="category",
)
op.drop_constraint(
"fk_categorized_name_categories_available",
"categorized",
schema="transactions",
type_="foreignkey",
)
op.create_foreign_key(
op.f("fk_categorized_name_available"),
"categorized",
"available",
["name"],
["name"],
source_schema="transactions",
referent_schema="category",
ondelete="CASCADE",
)
op.drop_constraint(
"fk_originals_bank_banks",
"originals",
schema="transactions",
type_="foreignkey",
)
op.create_foreign_key(
op.f("fk_originals_bank_banks"),
"originals",
"banks",
["bank"],
["name"],
source_schema="transactions",
referent_schema="bank",
)
op.drop_constraint(
"fk_tags_tag_tags_available", "tags", schema="transactions", type_="foreignkey"
)
op.create_foreign_key(
op.f("fk_tags_tag_available"),
"tags",
"available",
["tag"],
["name"],
source_schema="transactions",
referent_schema="tag",
)
op.drop_table("categories_schedules", schema="transactions")
op.drop_table("categories_rules", schema="transactions")
op.drop_table("categories_available", schema="transactions")
op.drop_table("tag_rules", schema="transactions")
op.drop_table("nordigen", schema="transactions")
op.drop_table("tags_available", schema="transactions")
op.drop_table("banks", schema="transactions")
op.drop_table("categories_selector", schema="transactions")
op.drop_table("categories_groups", schema="transactions")
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(
op.f("fk_tags_tag_available"), "tags", schema="transactions", type_="foreignkey"
)
op.create_foreign_key(
"fk_tags_tag_tags_available",
"tags",
"tags_available",
["tag"],
["name"],
source_schema="transactions",
referent_schema="transactions",
)
op.drop_constraint(
op.f("fk_originals_bank_banks"),
"originals",
schema="transactions",
type_="foreignkey",
)
op.create_foreign_key(
"fk_originals_bank_banks",
"originals",
"banks",
["bank"],
["name"],
source_schema="transactions",
referent_schema="transactions",
)
op.drop_constraint(
op.f("fk_categorized_name_available"),
"categorized",
schema="transactions",
type_="foreignkey",
)
op.create_foreign_key(
"fk_categorized_name_categories_available",
"categorized",
"categories_available",
["name"],
["name"],
source_schema="transactions",
referent_schema="transactions",
ondelete="CASCADE",
)
op.create_table(
"categories_groups",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint("name", name="pk_categories_groups"),
schema="transactions",
postgresql_ignore_search_path=False,
)
op.create_table(
"categories_selector",
sa.Column("id", sa.BIGINT(), autoincrement=False, nullable=False),
sa.Column(
"selector",
postgresql.ENUM(
"unknown",
"nullifier",
"vacations",
"rules",
"algorithm",
"manual",
name="selector",
schema="transactions",
),
autoincrement=False,
nullable=False,
),
sa.ForeignKeyConstraint(
["id"],
["transactions.categorized.id"],
name="fk_categories_selector_id_categorized",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name="pk_categories_selector"),
schema="transactions",
)
op.create_table(
"banks",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column("BIC", sa.VARCHAR(length=8), autoincrement=False, nullable=False),
sa.Column(
"type",
postgresql.ENUM(
"checking",
"savings",
"investment",
"mealcard",
"VISA",
"MASTERCARD",
name="accounttype",
schema="transactions",
),
autoincrement=False,
nullable=False,
),
sa.PrimaryKeyConstraint("BIC", "type", name="pk_banks"),
sa.UniqueConstraint("name", name="uq_banks_name"),
schema="transactions",
postgresql_ignore_search_path=False,
)
op.create_table(
"tags_available",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint("name", name="pk_tags_available"),
schema="transactions",
postgresql_ignore_search_path=False,
)
op.create_table(
"nordigen",
sa.Column("name", sa.TEXT(), autoincrement=False, nullable=False),
sa.Column("bank_id", sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column("requisition_id", sa.VARCHAR(), autoincrement=False, nullable=True),
sa.Column("invert", sa.BOOLEAN(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(
["name"], ["transactions.banks.name"], name="fk_nordigen_name_banks"
),
sa.PrimaryKeyConstraint("name", name="pk_nordigen"),
schema="transactions",
)
op.create_table(
"tag_rules",
sa.Column(
"id",
sa.BIGINT(),
server_default=sa.text(
"nextval('transactions.tag_rules_id_seq'::regclass)"
),
autoincrement=True,
nullable=False,
),
sa.Column("tag", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(
["id"],
["transactions.rules.id"],
name="fk_tag_rules_id_rules",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["tag"],
["transactions.tags_available.name"],
name="fk_tag_rules_tag_tags_available",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name="pk_tag_rules"),
schema="transactions",
)
op.create_table(
"categories_rules",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column(
"id",
sa.BIGINT(),
server_default=sa.text(
"nextval('transactions.categories_rules_id_seq'::regclass)"
),
autoincrement=True,
nullable=False,
),
sa.ForeignKeyConstraint(
["id"],
["transactions.rules.id"],
name="fk_categories_rules_id_rules",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["name"],
["transactions.categories_available.name"],
name="fk_categories_rules_name_categories_available",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name="pk_categories_rules"),
schema="transactions",
)
op.create_table(
"categories_available",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column("group", sa.VARCHAR(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(
["group"],
["transactions.categories_groups.name"],
name="fk_categories_available_group_categories_groups",
),
sa.PrimaryKeyConstraint("name", name="pk_categories_available"),
schema="transactions",
postgresql_ignore_search_path=False,
)
op.create_table(
"categories_schedules",
sa.Column("name", sa.VARCHAR(), autoincrement=False, nullable=False),
sa.Column(
"period",
postgresql.ENUM(
"daily",
"weekly",
"monthly",
"yearly",
name="period",
schema="transactions",
),
autoincrement=False,
nullable=True,
),
sa.Column(
"period_multiplier", sa.INTEGER(), autoincrement=False, nullable=True
),
sa.Column("amount", sa.INTEGER(), autoincrement=False, nullable=True),
sa.ForeignKeyConstraint(
["name"],
["transactions.categories_available.name"],
name="fk_categories_schedules_name_categories_available",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("name", name="pk_categories_schedules"),
schema="transactions",
)
op.drop_table("selector", schema="category")
op.drop_table("schedules", schema="category")
op.drop_table("rules", schema="category")
op.drop_table("rules", schema="tag")
op.drop_table("available", schema="category")
op.drop_table("nordigen", schema="bank")
op.drop_table("available", schema="tag")
op.drop_table("groups", schema="category")
op.drop_table("banks", schema="bank")
# ### end Alembic commands ###

15
pfbudget/__main__.py
View File

@ -232,20 +232,7 @@ if __name__ == "__main__":
pfbudget.t.Link(args["original"][0], link) for link in args["links"]
]
case (
pfbudget.Operation.Export
| pfbudget.Operation.Import
| pfbudget.Operation.ExportBanks
| pfbudget.Operation.ImportBanks
| pfbudget.Operation.ExportCategoryRules
| pfbudget.Operation.ImportCategoryRules
| pfbudget.Operation.ExportTagRules
| pfbudget.Operation.ImportTagRules
| pfbudget.Operation.ExportCategories
| pfbudget.Operation.ImportCategories
| pfbudget.Operation.ExportCategoryGroups
| pfbudget.Operation.ImportCategoryGroups
):
case pfbudget.Operation.Export | pfbudget.Operation.Import | pfbudget.Operation.ExportCategoryRules | pfbudget.Operation.ImportCategoryRules | pfbudget.Operation.ExportTagRules | pfbudget.Operation.ImportTagRules:
keys = {"file"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"

24
pfbudget/cli/runnable.py
View File

@ -225,14 +225,6 @@ def bank(parser: argparse.ArgumentParser):
nordigen(commands.add_parser("nordigen"))
export = commands.add_parser("export")
export.set_defaults(op=Operation.ExportBanks)
export_args(export)
pimport = commands.add_parser("import")
pimport.set_defaults(op=Operation.ImportBanks)
export_args(pimport)
def nordigen(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
@ -286,14 +278,6 @@ def category(parser: argparse.ArgumentParser):
group = commands.add_parser("group")
category_group(group)
export = commands.add_parser("export")
export.set_defaults(op=Operation.ExportCategories)
export_args(export)
pimport = commands.add_parser("import")
pimport.set_defaults(op=Operation.ImportCategories)
export_args(pimport)
def category_group(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
@ -306,14 +290,6 @@ def category_group(parser: argparse.ArgumentParser):
remove.set_defaults(op=Operation.GroupRemove)
remove.add_argument("group", nargs="+", type=str)
export = commands.add_parser("export")
export.set_defaults(op=Operation.ExportCategoryGroups)
export_args(export)
pimport = commands.add_parser("import")
pimport.set_defaults(op=Operation.ImportCategoryGroups)
export_args(pimport)
def category_rule(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)

6
pfbudget/common/types.py
View File

@ -37,16 +37,10 @@ class Operation(Enum):
NordigenCountryBanks = auto()
Export = auto()
Import = auto()
ExportBanks = auto()
ImportBanks = auto()
ExportCategoryRules = auto()
ImportCategoryRules = auto()
ExportTagRules = auto()
ImportTagRules = auto()
ExportCategories = auto()
ImportCategories = auto()
ExportCategoryGroups = auto()
ImportCategoryGroups = auto()
class TransactionError(Exception):

84
pfbudget/core/manager.py
View File

@ -11,7 +11,6 @@ from pfbudget.db.model import (
Category,
CategoryGroup,
CategoryRule,
CategorySchedule,
CategorySelector,
Link,
MoneyTransaction,
@ -214,26 +213,16 @@ class Manager:
transactions.append(transaction)
if self.certify(transactions):
if (
len(transactions) > 0
and input(
f"{transactions[:5]}\nDoes the import seem correct? (y/n)"
)
== "y"
):
with self.db.session() as session:
session.add(transactions)
case Operation.ExportBanks:
with self.db.session() as session:
self.dump(params[0], session.get(Bank))
case Operation.ImportBanks:
banks = []
for row in self.load(params[0]):
bank = Bank(row["name"], row["BIC"], row["type"])
if row["nordigen"]:
bank.nordigen = Nordigen(**row["nordigen"])
banks.append(bank)
if self.certify(banks):
with self.db.session() as session:
session.add(banks)
case Operation.ExportCategoryRules:
with self.db.session() as session:
self.dump(params[0], session.get(CategoryRule))
@ -241,7 +230,11 @@ class Manager:
case Operation.ImportCategoryRules:
rules = [CategoryRule(**row) for row in self.load(params[0])]
if self.certify(rules):
if (
len(rules) > 0
and input(f"{rules[:5]}\nDoes the import seem correct? (y/n)")
== "y"
):
with self.db.session() as session:
session.add(rules)
@ -252,64 +245,25 @@ class Manager:
case Operation.ImportTagRules:
rules = [TagRule(**row) for row in self.load(params[0])]
if self.certify(rules):
if (
len(rules) > 0
and input(f"{rules[:5]}\nDoes the import seem correct? (y/n)")
== "y"
):
with self.db.session() as session:
session.add(rules)
case Operation.ExportCategories:
with self.db.session() as session:
self.dump(params[0], session.get(Category))
case Operation.ImportCategories:
# rules = [Category(**row) for row in self.load(params[0])]
categories = []
for row in self.load(params[0]):
category = Category(row["name"], row["group"])
if len(row["rules"]) > 0:
# Only category rules could have been created with a rule
rules = row["rules"]
for rule in rules:
del rule["type"]
category.rules = set(CategoryRule(**rule) for rule in rules)
if row["schedule"]:
category.schedule = CategorySchedule(**row["schedule"])
categories.append(category)
if self.certify(categories):
with self.db.session() as session:
session.add(categories)
case Operation.ExportCategoryGroups:
with self.db.session() as session:
self.dump(params[0], session.get(CategoryGroup))
case Operation.ImportCategoryGroups:
groups = [CategoryGroup(**row) for row in self.load(params[0])]
if self.certify(groups):
with self.db.session() as session:
session.add(groups)
def parse(self, filename: Path, args: dict):
return parse_data(filename, args)
@staticmethod
def dump(fn, sequence):
def dump(self, fn, sequence):
with open(fn, "wb") as f:
pickle.dump([e.format for e in sequence], f)
@staticmethod
def load(fn):
def load(self, fn):
with open(fn, "rb") as f:
return pickle.load(f)
@staticmethod
def certify(imports: list) -> bool:
if input(f"{imports[:10]}\nDoes the import seem correct? (y/n)") == "y":
return True
return False
@property
def db(self) -> DbClient:
return DbClient(self._db, self._verbosity > 2)

88
pfbudget/db/model.py
View File

@ -24,15 +24,15 @@ from sqlalchemy.orm import (
class Base(MappedAsDataclass, DeclarativeBase):
__table_args__ = {"schema": "transactions"}
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",
},
}
)
@ -57,8 +57,7 @@ class Export:
raise NotImplementedError
class Bank(Base, Export):
__table_args__ = {"schema": "bank"}
class Bank(Base):
__tablename__ = "banks"
name: Mapped[str] = mapped_column(unique=True)
@ -67,15 +66,6 @@ class Bank(Base, Export):
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))]
@ -84,7 +74,7 @@ money = Annotated[decimal.Decimal, mapped_column(Numeric(16, 2))]
class Transaction(Base, Export):
__tablename__ = "transactions"
__tablename__ = "originals"
id: Mapped[idpk] = mapped_column(init=False)
date: Mapped[dt.date]
@ -147,20 +137,14 @@ class SplitTransaction(Transaction):
return super().format | dict(original=self.original)
class CategoryGroup(Base, Export):
__table_args__ = {"schema": "category"}
__tablename__ = "groups"
class CategoryGroup(Base):
__tablename__ = "categories_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"
class Category(Base):
__tablename__ = "categories_available"
name: Mapped[str] = mapped_column(primary_key=True)
group: Mapped[Optional[str]] = mapped_column(
@ -177,15 +161,6 @@ class Category(Base, Export):
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,
@ -215,8 +190,7 @@ class Note(Base):
note: Mapped[str]
class Nordigen(Base, Export):
__table_args__ = {"schema": "bank"}
class Nordigen(Base):
__tablename__ = "nordigen"
name: Mapped[bankfk] = mapped_column(primary_key=True)
@ -224,19 +198,9 @@ class Nordigen(Base, Export):
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"
__tablename__ = "tags_available"
name: Mapped[str] = mapped_column(primary_key=True)
@ -246,7 +210,7 @@ class Tag(Base):
class TransactionTag(Base, Export):
__tablename__ = "tagged"
__tablename__ = "tags"
id: Mapped[idfk] = mapped_column(primary_key=True, init=False)
tag: Mapped[str] = mapped_column(ForeignKey(Tag.name), primary_key=True)
@ -259,7 +223,7 @@ class TransactionTag(Base, Export):
return hash(self.id)
class Selector_T(enum.Enum):
class Selector(enum.Enum):
unknown = enum.auto()
nullifier = enum.auto()
vacations = enum.auto()
@ -269,13 +233,13 @@ class Selector_T(enum.Enum):
categoryselector = Annotated[
Selector_T,
mapped_column(Enum(Selector_T, inherit_schema=True), default=Selector_T.unknown),
Selector,
mapped_column(Enum(Selector, inherit_schema=True), default=Selector.unknown),
]
class CategorySelector(Base, Export):
__tablename__ = "selector"
__tablename__ = "categories_selector"
id: Mapped[int] = mapped_column(
BigInteger,
@ -297,27 +261,17 @@ class Period(enum.Enum):
yearly = "yearly"
scheduleperiod = Annotated[Selector_T, mapped_column(Enum(Period, inherit_schema=True))]
scheduleperiod = Annotated[Selector, mapped_column(Enum(Period, inherit_schema=True))]
class CategorySchedule(Base, Export):
__table_args__ = {"schema": "category"}
__tablename__ = "schedules"
class CategorySchedule(Base):
__tablename__ = "categories_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"
@ -380,8 +334,7 @@ class Rule(Base, Export):
class CategoryRule(Rule):
__table_args__ = {"schema": "category"}
__tablename__ = "rules"
__tablename__ = "categories_rules"
id: Mapped[int] = mapped_column(
BigInteger,
@ -404,8 +357,7 @@ class CategoryRule(Rule):
class TagRule(Rule):
__table_args__ = {"schema": "tag"}
__tablename__ = "rules"
__tablename__ = "tag_rules"
id: Mapped[int] = mapped_column(
BigInteger,