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

Compare commits

base: personal-finance:1d256d7def31748920199d739cdd7c9c443c36ab
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1
..
compare: personal-finance:d321481e29a3c98e148021d81ac0d735bf93fe07
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1

No commits in common. "1d256d7def31748920199d739cdd7c9c443c36ab" and "d321481e29a3c98e148021d81ac0d735bf93fe07" have entirely different histories.
1d256d7def ... d321481e29

13 changed files with 116 additions and 697 deletions
Show Stats Expand all files Collapse all files

32
alembic/versions/0ce89e987770_regex_rule.py
View File

@ -1,32 +0,0 @@
"""Regex rule
Revision ID: 0ce89e987770
Revises: 7adf89ec8d14
Create Date: 2022-12-10 14:00:49.418494+00:00
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "0ce89e987770"
down_revision = "7adf89ec8d14"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.add_column(
"categories_rules",
sa.Column("regex", sa.String(), nullable=True),
schema="transactions",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column("categories_rules", "regex", schema="transactions")
# ### end Alembic commands ###

49
alembic/versions/753c0bfb2062_rule_min_max.py
View File

@ -1,49 +0,0 @@
"""Rule min/max
Revision ID: 753c0bfb2062
Revises: e36e6321568e
Create Date: 2022-12-18 00:24:03.861461+00:00
"""
from alembic import op
# revision identifiers, used by Alembic.
revision = "753c0bfb2062"
down_revision = "e36e6321568e"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column(
"categories_rules",
"min_amount",
new_column_name="min",
schema="transactions",
)
op.alter_column(
"categories_rules",
"max_amount",
new_column_name="max",
schema="transactions",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column(
"categories_rules",
"min",
new_column_name="min_amount",
schema="transactions",
)
op.alter_column(
"categories_rules",
"max",
new_column_name="max_amount",
schema="transactions",
)
# ### end Alembic commands ###

46
alembic/versions/8cc9870b0d74_links.py
View File

@ -1,46 +0,0 @@
"""Links
Revision ID: 8cc9870b0d74
Revises: a910e1b2214d
Create Date: 2022-12-19 22:10:25.136479+00:00
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "8cc9870b0d74"
down_revision = "a910e1b2214d"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"links",
sa.Column("original", sa.BigInteger(), nullable=False),
sa.Column("link", sa.BigInteger(), nullable=False),
sa.ForeignKeyConstraint(
["link"],
["transactions.originals.id"],
name=op.f("fk_links_link_originals"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["original"],
["transactions.originals.id"],
name=op.f("fk_links_original_originals"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("original", "link", name=op.f("pk_links")),
schema="transactions",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("links", schema="transactions")
# ### end Alembic commands ###

68
alembic/versions/9028b0f3b985_tag_rules.py
View File

@ -1,68 +0,0 @@
"""Tag rules
Revision ID: 9028b0f3b985
Revises: 753c0bfb2062
Create Date: 2022-12-18 22:53:13.334046+00:00
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "9028b0f3b985"
down_revision = "753c0bfb2062"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"tags_available",
sa.Column("name", sa.String(), nullable=False),
sa.PrimaryKeyConstraint("name", name=op.f("pk_tags_available")),
schema="transactions",
)
op.create_table(
"tag_rules",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column("tag", sa.String(), nullable=False),
sa.Column("date", sa.Date(), nullable=True),
sa.Column("description", sa.String(), nullable=True),
sa.Column("regex", sa.String(), nullable=True),
sa.Column("bank", sa.String(), nullable=True),
sa.Column("min", sa.Numeric(precision=16, scale=2), nullable=True),
sa.Column("max", sa.Numeric(precision=16, scale=2), nullable=True),
sa.ForeignKeyConstraint(
["tag"],
["transactions.tags_available.name"],
name=op.f("fk_tag_rules_tag_tags_available"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_tag_rules")),
schema="transactions",
)
op.create_foreign_key(
op.f("fk_tags_tag_tags_available"),
"tags",
"tags_available",
["tag"],
["name"],
source_schema="transactions",
referent_schema="transactions",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(
op.f("fk_tags_tag_tags_available"),
"tags",
schema="transactions",
type_="foreignkey",
)
op.drop_table("tag_rules", schema="transactions")
op.drop_table("tags_available", schema="transactions")
# ### end Alembic commands ###

56
alembic/versions/a910e1b2214d_rule_inheritance.py
View File

@ -1,56 +0,0 @@
"""Rule inheritance
Revision ID: a910e1b2214d
Revises: 9028b0f3b985
Create Date: 2022-12-19 20:48:04.682812+00:00
"""
from alembic import op
# revision identifiers, used by Alembic.
revision = "a910e1b2214d"
down_revision = "9028b0f3b985"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(
"fk_categorized_name_categories_available",
"categorized",
schema="transactions",
type_="foreignkey",
)
op.create_foreign_key(
op.f("fk_categorized_name_categories_available"),
"categorized",
"categories_available",
["name"],
["name"],
source_schema="transactions",
referent_schema="transactions",
ondelete="CASCADE",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(
op.f("fk_categorized_name_categories_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",
)
# ### end Alembic commands ###

58
alembic/versions/e36e6321568e_rules_min_max_money.py
View File

@ -1,58 +0,0 @@
"""Rules min/max money
Revision ID: e36e6321568e
Revises: 0ce89e987770
Create Date: 2022-12-10 18:55:07.149010+00:00
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "e36e6321568e"
down_revision = "0ce89e987770"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column(
"categories_rules",
"min_amount",
existing_type=sa.DOUBLE_PRECISION(precision=53),
type_=sa.Numeric(precision=16, scale=2),
existing_nullable=True,
schema="transactions",
)
op.alter_column(
"categories_rules",
"max_amount",
existing_type=sa.DOUBLE_PRECISION(precision=53),
type_=sa.Numeric(precision=16, scale=2),
existing_nullable=True,
schema="transactions",
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column(
"categories_rules",
"max_amount",
existing_type=sa.Numeric(precision=16, scale=2),
type_=sa.DOUBLE_PRECISION(precision=53),
existing_nullable=True,
schema="transactions",
)
op.alter_column(
"categories_rules",
"min_amount",
existing_type=sa.Numeric(precision=16, scale=2),
type_=sa.DOUBLE_PRECISION(precision=53),
existing_nullable=True,
schema="transactions",
)
# ### end Alembic commands ###

96
pfbudget/__main__.py
View File

@ -6,26 +6,24 @@ if __name__ == "__main__":
args = vars(argparser.parse_args())
assert "op" in args, "No pfbudget.Operation selected"
op: pfbudget.Operation = args.pop("op")
op: pfbudget.Operation = args["op"]
assert "database" in args, "No database selected"
db = args.pop("database")
assert "verbose" in args, "No verbose level specified"
verbosity = args.pop("verbose")
db = args["database"]
params = None
match (op):
case pfbudget.Operation.CategoryAdd:
assert args.keys() >= {"category", "group"}, "argparser ill defined"
params = [
pfbudget.types.Category(cat, args["group"]) for cat in args["category"]
pfbudget.types.Category(cat, args["group"][0])
for cat in args["category"]
]
case pfbudget.Operation.CategoryUpdate:
assert args.keys() >= {"category", "group"}, "argparser ill defined"
params = [pfbudget.types.Category(cat) for cat in args["category"]]
params.append(args["group"])
params.append(args["group"][0])
case pfbudget.Operation.CategoryRemove:
assert "category" in args, "argparser ill defined"
@ -45,7 +43,7 @@ if __name__ == "__main__":
for cat in args["category"]
]
case pfbudget.Operation.RuleAdd:
case pfbudget.Operation.CategoryRule:
assert args.keys() >= {
"category",
"date",
@ -60,7 +58,6 @@ if __name__ == "__main__":
cat,
args["date"][0] if args["date"] else None,
args["description"][0] if args["description"] else None,
args["regex"][0] if args["regex"] else None,
args["bank"][0] if args["bank"] else None,
args["min"][0] if args["min"] else None,
args["max"][0] if args["max"] else None,
@ -68,79 +65,6 @@ if __name__ == "__main__":
for cat in args["category"]
]
case pfbudget.Operation.RuleRemove | pfbudget.Operation.TagRuleRemove:
assert args.keys() >= {"id"}, "argparser ill defined"
params = args["id"]
case pfbudget.Operation.RuleModify:
assert args.keys() >= {
"id",
"category",
"date",
"description",
"bank",
"min",
"max",
"remove",
}, "argparser ill defined"
nargs_1 = ["category", "date", "description", "regex", "bank", "min", "max"]
params = []
for id in args["id"]:
param = {"id": id}
param |= {k: v[0] for k, v in args.items() if k in nargs_1 and args[k]}
param |= {k: None for k in args["remove"] if k in nargs_1}
params.append(param)
case pfbudget.Operation.TagAdd:
assert args.keys() >= {"tag"}, "argparser ill defined"
params = [pfbudget.types.Tag(tag) for tag in args["tag"]]
case pfbudget.Operation.TagRuleAdd:
assert args.keys() >= {
"tag",
"date",
"description",
"bank",
"min",
"max",
}, "argparser ill defined"
params = [
pfbudget.types.TagRule(
tag,
args["date"][0] if args["date"] else None,
args["description"][0] if args["description"] else None,
args["regex"][0] if args["regex"] else None,
args["bank"][0] if args["bank"] else None,
args["min"][0] if args["min"] else None,
args["max"][0] if args["max"] else None,
)
for tag in args["tag"]
]
case pfbudget.Operation.TagRuleModify:
assert args.keys() >= {
"id",
"tag",
"date",
"description",
"bank",
"min",
"max",
"remove",
}, "argparser ill defined"
nargs_1 = ["tag", "date", "description", "regex", "bank", "min", "max"]
params = []
for id in args["id"]:
param = {"id": id}
param |= {k: v[0] for k, v in args.items() if k in nargs_1 and args[k]}
param |= {k: None for k in args["remove"] if k in nargs_1}
params.append(param)
case pfbudget.Operation.GroupAdd:
assert "group" in args, "argparser ill defined"
params = [pfbudget.types.CategoryGroup(group) for group in args["group"]]
@ -149,10 +73,4 @@ if __name__ == "__main__":
assert "group" in args, "argparser ill defined"
params = [pfbudget.types.CategoryGroup(group) for group in args["group"]]
case pfbudget.Operation.Forge | pfbudget.Operation.Dismantle:
assert args.keys() >= {"original", "links"}, "argparser ill defined"
params = [
pfbudget.types.Link(args["original"][0], link) for link in args["links"]
]
pfbudget.Manager(db, verbosity, args).action(op, params)
pfbudget.Manager(db, args).action(op, params)

100
pfbudget/cli/runnable.py
View File

@ -1,11 +1,12 @@
from pathlib import Path
import argparse
import datetime as dt
import decimal
import re
from pfbudget.common.types import Operation
from pfbudget.core.categories import categorize_data
from pfbudget.db.model import Period
from pfbudget.input.json import JsonParser
from pfbudget.input.nordigen import NordigenInput
from pfbudget.db.sqlite import DatabaseClient
import pfbudget.reporting.graph
@ -38,7 +39,12 @@ def argparser() -> argparse.ArgumentParser:
help="select current database",
default=DEFAULT_DB,
)
universal.add_argument("-v", "--verbose", action="count", default=0)
universal.add_argument(
"-q", "--quiet", action="store_true", help="reduces the amount of verbose"
)
universal.add_argument(
"-v", "--verbose", action="store_true", help="increases the amount of verbose"
)
period = argparse.ArgumentParser(add_help=False).add_mutually_exclusive_group()
period.add_argument(
@ -248,12 +254,6 @@ def argparser() -> argparse.ArgumentParser:
category_parser = subparsers.add_parser("category", parents=[universal])
category(category_parser, universal)
# Tag
tags(subparsers.add_parser("tag", parents=[universal]), universal)
# Link
link(subparsers.add_parser("link"))
return parser
@ -341,7 +341,13 @@ def category(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser
schedule.add_argument("--frequency", nargs=1, default=[1], type=int)
rule = commands.add_parser("rule", parents=[universal])
category_rule(rule, universal)
rule.set_defaults(op=Operation.CategoryRule)
rule.add_argument("category", nargs="+", type=str)
rule.add_argument("--date", nargs=1, type=dt.date.fromisoformat)
rule.add_argument("--description", nargs=1, type=str)
rule.add_argument("--bank", nargs=1, type=str)
rule.add_argument("--min", nargs=1, type=float)
rule.add_argument("--max", nargs=1, type=float)
group = commands.add_parser("group", parents=[universal])
category_group(group, universal)
@ -359,82 +365,6 @@ def category_group(parser: argparse.ArgumentParser, universal: argparse.Argument
remove.add_argument("group", nargs="+", type=str)
def category_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add.set_defaults(op=Operation.RuleAdd)
add.add_argument("category", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove", parents=[universal])
remove.set_defaults(op=Operation.RuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify", parents=[universal])
modify.set_defaults(op=Operation.RuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--category", nargs=1, type=str)
rules(modify)
def tags(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add.set_defaults(op=Operation.TagAdd)
add.add_argument("tag", nargs="+", type=str)
remove = commands.add_parser("remove", parents=[universal])
remove.set_defaults(op=Operation.TagRemove)
remove.add_argument("tag", nargs="+", type=str)
rule = commands.add_parser("rule", parents=[universal])
tag_rule(rule, universal)
def tag_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add.set_defaults(op=Operation.TagRuleAdd)
add.add_argument("tag", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove", parents=[universal])
remove.set_defaults(op=Operation.TagRuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify", parents=[universal])
modify.set_defaults(op=Operation.TagRuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--tag", nargs=1, type=str)
rules(modify)
def rules(parser: argparse.ArgumentParser):
parser.add_argument("--date", nargs=1, type=dt.date.fromisoformat)
parser.add_argument("--description", nargs=1, type=str)
parser.add_argument("--regex", nargs=1, type=str)
parser.add_argument("--bank", nargs=1, type=str)
parser.add_argument("--min", nargs=1, type=decimal.Decimal)
parser.add_argument("--max", nargs=1, type=decimal.Decimal)
def link(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
forge = commands.add_parser("forge")
forge.set_defaults(op=Operation.Forge)
forge.add_argument("original", nargs=1, type=int)
forge.add_argument("links", nargs="+", type=int)
dismantle = commands.add_parser("dismantle")
dismantle.set_defaults(op=Operation.Dismantle)
dismantle.add_argument("original", nargs=1, type=int)
dismantle.add_argument("links", nargs="+", type=int)
def run():
args = vars(argparser().parse_args())
assert "op" in args, "No operation selected"

11
pfbudget/common/types.py
View File

@ -17,18 +17,9 @@ class Operation(Enum):
CategoryUpdate = auto()
CategoryRemove = auto()
CategorySchedule = auto()
RuleAdd = auto()
RuleRemove = auto()
RuleModify = auto()
CategoryRule = auto()
GroupAdd = auto()
GroupRemove = auto()
TagAdd = auto()
TagRemove = auto()
TagRuleAdd = auto()
TagRuleRemove = auto()
TagRuleModify = auto()
Forge = auto()
Dismantle = auto()
class TransactionError(Exception):

76
pfbudget/core/categorizer.py
View File

@ -2,10 +2,8 @@ from pfbudget.db.model import (
Category,
CategorySelector,
Selector,
Tag,
Transaction,
TransactionCategory,
TransactionTag,
)
from datetime import timedelta
@ -17,12 +15,7 @@ class Categorizer:
def __init__(self):
self.options["null_days"] = 4
def categorize(
self,
transactions: list[Transaction],
categories: list[Category],
tags: list[Tag],
):
def categorize(self, transactions: list[Transaction], categories: list[Category]):
"""Overarching categorization tool
Receives a list of transactions (by ref) and updates their category
@ -32,8 +25,7 @@ class Categorizer:
"""
self._nullify(transactions)
self._rule_based_categories(transactions, categories)
self._rule_based_tags(transactions, tags)
self._rules(transactions, categories)
def _nullify(self, transactions: list[Transaction]):
count = 0
@ -65,57 +57,27 @@ class Categorizer:
print(f"Nullified {count} transactions")
def _rule_based_categories(
self, transactions: list[Transaction], categories: list[Category]
):
d = {}
def _rules(self, transactions: list[Transaction], categories: list[Category]):
for category in [c for c in categories if c.rules]:
for rule in category.rules:
# for transaction in [t for t in transactions if not t.category]:
for transaction in [
t
for t in transactions
if not t.category or t.category.name != "null"
]:
if not rule.matches(transaction):
continue
for transaction in [t for t in transactions if not t.category]:
if rule.date:
if rule.date < transaction.date:
continue
if rule.description:
if rule.description not in transaction.description:
continue
if rule.bank:
if rule.bank != transaction.bank:
continue
if rule.min_amount:
if rule.min_amount > transaction.amount:
continue
if rule.max_amount:
if rule.max_amount <= transaction.amount:
continue
# passed all conditions, assign category
transaction.category = TransactionCategory(
category.name, CategorySelector(Selector.rules)
)
if rule in d:
d[rule] += 1
else:
d[rule] = 1
for k, v in d.items():
print(f"{v}: {k}")
def _rule_based_tags(self, transactions: list[Transaction], tags: list[Tag]):
d = {}
for tag in [t for t in tags if t.rules]:
for rule in tag.rules:
# for transaction in [t for t in transactions if not t.category]:
for transaction in [
t
for t in transactions
if tag.name not in [tag.tag for tag in t.tags]
]:
if not rule.matches(transaction):
continue
if not transaction.tags:
transaction.tags = {TransactionTag(tag.name)}
else:
transaction.tags.add(TransactionTag(tag.name))
if rule in d:
d[rule] += 1
else:
d[rule] = 1
for k, v in d.items():
print(f"{v}: {k}")

65
pfbudget/core/manager.py
View File

@ -2,14 +2,7 @@ from pfbudget.input.input import Input
from pfbudget.input.nordigen import NordigenClient
from pfbudget.input.parsers import parse_data
from pfbudget.db.client import DbClient
from pfbudget.db.model import (
Category,
CategoryGroup,
CategoryRule,
CategorySchedule,
Tag,
TagRule,
)
from pfbudget.db.model import Category, CategoryGroup, CategoryRule, CategorySchedule
from pfbudget.common.types import Operation
from pfbudget.core.categorizer import Categorizer
from pfbudget.utils import convert
@ -18,16 +11,13 @@ from pfbudget.cli.runnable import download, parse
class Manager:
def __init__(self, db: str, verbosity: int = 0, args: dict = {}):
def __init__(self, db: str, args: dict):
self._args = args
print(args)
self._db = db
self._verbosity = verbosity
def action(self, op: Operation, params: list):
if self._verbosity > 0:
print(f"op={op}, params={params}")
match (op):
case Operation.Init:
pass
@ -37,13 +27,11 @@ class Manager:
case Operation.Download:
# TODO this is a monstrosity, remove when possible
download(self, self.args)
case Operation.Categorize:
with self.db.session() as session:
uncategorized = session.uncategorized()
categories = session.categories()
tags = session.tags()
Categorizer().categorize(uncategorized, categories, tags)
Categorizer().categorize(uncategorized, categories)
case Operation.Register:
# self._db = DbClient(args["database"])
@ -61,9 +49,9 @@ class Manager:
self.args["name"], self.args["country"]
)
case Operation.CategoryAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
case Operation.CategoryAdd:
with self.db.session() as session:
session.add(params)
session.addcategories(params)
case Operation.CategoryUpdate:
with self.db.session() as session:
@ -71,49 +59,26 @@ class Manager:
case Operation.CategoryRemove:
with self.db.session() as session:
session.remove_by_name(Category, params)
session.removecategories(params)
case Operation.CategorySchedule:
with self.db.session() as session:
session.updateschedules(params)
case Operation.RuleRemove:
assert all(isinstance(param, int) for param in params)
case Operation.CategoryRule:
with self.db.session() as session:
session.remove_by_id(CategoryRule, params)
case Operation.TagRemove:
with self.db.session() as session:
session.remove_by_name(Tag, params)
case Operation.TagRuleRemove:
assert all(isinstance(param, int) for param in params)
with self.db.session() as session:
session.remove_by_id(TagRule, params)
case Operation.RuleModify | Operation.TagRuleModify:
assert all(isinstance(param, dict) for param in params)
with self.db.session() as session:
session.updaterules(params)
session.addrules(params)
case Operation.GroupAdd:
with self.db.session() as session:
session.add(CategoryGroup(params))
for group in self.args["group"]:
session.addcategorygroup(CategoryGroup(name=group))
case Operation.GroupRemove:
assert all(isinstance(param, CategoryGroup) for param in params)
with self.db.session() as session:
session.remove_by_name(CategoryGroup, params)
case Operation.Forge:
with self.db.session() as session:
session.add(params)
case Operation.Dismantle:
with self.db.session() as session:
original = params[0].original
links = [link.link for link in params]
session.remove_links(original, links)
session.removecategorygroup(
[CategoryGroup(name=group) for group in self.args["group"]]
)
# def init(self):
# client = DatabaseClient(self.__db)
@ -154,7 +119,7 @@ class Manager:
@property
def db(self) -> DbClient:
return DbClient(self._db, self._verbosity > 0)
return DbClient(self._db, self.args["verbose"])
@db.setter
def db(self, url: str):

36
pfbudget/db/client.py
View File

@ -1,3 +1,4 @@
from copy import deepcopy
from dataclasses import asdict
from sqlalchemy import create_engine, delete, select, update
from sqlalchemy.dialects.postgresql import insert
@ -9,9 +10,6 @@ from pfbudget.db.model import (
CategoryGroup,
CategoryRule,
CategorySchedule,
Link,
Tag,
TagRule,
Transaction,
)
@ -83,11 +81,16 @@ class DbClient:
def commit(self):
self.__session.commit()
def add(self, rows: list):
self.__session.add_all(rows)
def add(self, transactions: list[Transaction]):
self.__session.add_all(transactions)
def remove_by_name(self, type: Category | Tag | Transaction, rows: list):
stmt = delete(type).where(type.name.in_([row.name for row in rows]))
def addcategories(self, category: list[Category]):
self.__session.add_all(category)
def removecategories(self, categories: list[Category]):
stmt = delete(Category).where(
Category.name.in_([cat.name for cat in categories])
)
self.__session.execute(stmt)
def updategroup(self, categories: list[Category], group: CategoryGroup):
@ -110,15 +113,16 @@ class DbClient:
)
self.__session.execute(stmt)
def remove_by_id(self, type: CategoryRule | TagRule, ids: list[int]):
stmt = delete(type).where(type.id.in_(ids))
self.__session.execute(stmt)
def addrules(self, rules: list[CategoryRule]):
self.__session.add_all(rules)
def updaterules(self, rules: list[dict]):
self.__session.execute(update(CategoryRule), rules)
def addcategorygroup(self, group: CategoryGroup):
self.__session.add(group)
def remove_links(self, original, links: list):
stmt = delete(Link).where(Link.original == original, Link.link.in_(link for link in links))
def removecategorygroup(self, groups: list[CategoryGroup]):
stmt = delete(CategoryGroup).where(
CategoryGroup.name.in_([grp.name for grp in groups])
)
self.__session.execute(stmt)
def uncategorized(self) -> list[Transaction]:
@ -129,9 +133,5 @@ class DbClient:
stmt = select(Category)
return self.__session.scalars(stmt).all()
def tags(self) -> list[Tag]:
stmt = select(Tag)
return self.__session.scalars(stmt).all()
def session(self) -> ClientSession:
return self.ClientSession(self.engine)

120
pfbudget/db/model.py
View File

@ -21,7 +21,6 @@ from decimal import Decimal
from typing import Annotated, Optional
import datetime as dt
import enum
import re
class Base(MappedAsDataclass, DeclarativeBase):
@ -63,25 +62,36 @@ class Bank(Base):
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, autoincrement=True)]
money = Annotated[Decimal, mapped_column(Numeric(16, 2))]
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(init=False)
id: Mapped[idpk] = mapped_column(autoincrement=True, init=False)
date: Mapped[dt.date]
description: Mapped[Optional[str]]
bank: Mapped[bankfk]
amount: Mapped[money]
category: Mapped[Optional[TransactionCategory]] = relationship()
note: Mapped[Optional[Note]] = relationship(back_populates="original")
tags: Mapped[Optional[set[TransactionTag]]] = relationship()
note: Mapped[Optional[Note]] = relationship(back_populates="original", default=None)
tags: Mapped[Optional[set[Tag]]] = relationship(
back_populates="original",
cascade="all, delete-orphan",
passive_deletes=True,
default=None,
)
def __repr__(self) -> str:
return f"Transaction(date={self.date}, description={self.description}, bank={self.bank}, amount={self.amount}, category={self.category})"
idfk = Annotated[
@ -104,7 +114,7 @@ class Category(Base):
)
rules: Mapped[Optional[set[CategoryRule]]] = relationship(
cascade="all, delete-orphan", passive_deletes=True, default_factory=set
cascade="all, delete-orphan", passive_deletes=True, default=None
)
schedule: Mapped[CategorySchedule] = relationship(
back_populates="category", default=None
@ -124,9 +134,9 @@ class TransactionCategory(Base):
__tablename__ = "categorized"
id: Mapped[idfk] = mapped_column(primary_key=True, init=False)
name: Mapped[catfk]
name: Mapped[str] = mapped_column(ForeignKey(Category.name))
selector: Mapped[CategorySelector] = relationship(cascade="all, delete-orphan")
selector: Mapped[CategorySelector] = relationship()
def __repr__(self) -> str:
return f"Category({self.name})"
@ -135,7 +145,7 @@ class TransactionCategory(Base):
class Note(Base):
__tablename__ = "notes"
id: Mapped[idfk] = mapped_column(primary_key=True, init=False)
id: Mapped[idfk] = mapped_column(primary_key=True)
note: Mapped[str]
original: Mapped[Transaction] = relationship(back_populates="note")
@ -151,22 +161,29 @@ class Nordigen(Base):
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_available"
name: Mapped[str] = mapped_column(primary_key=True)
rules: Mapped[Optional[set[TagRule]]] = relationship(
cascade="all, delete-orphan", passive_deletes=True, default_factory=set
)
class TransactionTag(Base):
__tablename__ = "tags"
id: Mapped[idfk] = mapped_column(primary_key=True, init=False)
tag: Mapped[str] = mapped_column(ForeignKey(Tag.name), primary_key=True)
id: Mapped[idfk] = mapped_column(primary_key=True)
tag: Mapped[str] = mapped_column(primary_key=True)
original: Mapped[Transaction] = relationship(back_populates="tags")
class CategoryRule(Base):
__tablename__ = "categories_rules"
id: Mapped[idpk] = mapped_column(autoincrement=True, init=False)
name: Mapped[catfk] = mapped_column()
date: Mapped[Optional[dt.date]] = mapped_column()
description: Mapped[Optional[str]] = mapped_column()
bank: Mapped[Optional[str]] = mapped_column()
min_amount: Mapped[Optional[float]] = mapped_column()
max_amount: Mapped[Optional[float]] = mapped_column()
def __hash__(self):
return hash(self.id)
@ -219,60 +236,5 @@ class CategorySchedule(Base):
category: Mapped[Category] = relationship(back_populates="schedule")
class Link(Base):
__tablename__ = "links"
original: Mapped[idfk] = mapped_column(primary_key=True)
link: Mapped[idfk] = mapped_column(primary_key=True)
class Rule:
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]]
def matches(self, transaction: Transaction) -> 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
class CategoryRule(Base, Rule):
__tablename__ = "categories_rules"
id: Mapped[idpk] = mapped_column(init=False)
name: Mapped[catfk]
def __hash__(self):
return hash(self.id)
class TagRule(Base, Rule):
__tablename__ = "tag_rules"
id: Mapped[idpk] = mapped_column(init=False)
tag: Mapped[str] = mapped_column(ForeignKey(Tag.name, ondelete="CASCADE"))
def __hash__(self):
return hash(self.id)
def __repr__(self) -> str:
return f"{self.name} schedule=Schedule(period={self.period}, multiplier={self.period_multiplier}, amount={self.amount})"