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

Fix download, bank, token and renew->eua ops

Browse Source 
- Update the download, token and eua cli with the new operations
- Change the bank and nordigen cli to be more in line with the other
  add/modify/remove operations. Also update manager logic.
- Fix some model.py leftovers
- Add __lt__ to Transaction to enable sorting
- Remove universal from child argparsers
This commit is contained in:
Luís Murta 2023-01-01 18:40:19 +00:00
parent 071711dbdb
commit 55a5b09c45
Signed by: satprog
GPG Key ID: 169EF1BBD7049F94
9 changed files with 283 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pfbudget/__init__.py
View File

@ -7,5 +7,6 @@ from pfbudget.core.categories import categorize_data
from pfbudget.core.manager import Manager
from pfbudget.cli.runnable import argparser
from pfbudget.input.parsers import parse_data
from pfbudget.utils.utils import parse_args_period
import pfbudget.db.model as types

92
pfbudget/__main__.py
View File

@ -16,6 +16,98 @@ if __name__ == "__main__":
params = None
match (op):
case pfbudget.Operation.RequisitionId:
assert args.keys() >= {"name", "country"}, "argparser ill defined"
params = [args["name"][0], args["country"][0]]
case pfbudget.Operation.Download:
assert args.keys() >= {
"id",
"name",
"all",
"interval",
"start",
"end",
"year",
}, "argparser ill defined"
start, end = pfbudget.parse_args_period(args)
params = [start, end]
case pfbudget.Operation.BankAdd:
assert args.keys() >= {
"bank",
"bic",
"type",
}, "argparser ill defined"
params = [
pfbudget.types.Bank(
args["bank"][0],
args["bic"][0],
args["type"][0],
)
]
case pfbudget.Operation.BankMod:
assert args.keys() >= {
"bank",
"bic",
"type",
"remove",
}, "argparser ill defined"
nargs_1 = ["bic", "type"]
param = {"name": args["bank"][0]}
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 = [param]
case pfbudget.Operation.BankDel:
assert len(args["bank"]) > 0, "argparser ill defined"
params = args["bank"]
case pfbudget.Operation.NordigenAdd:
assert args.keys() >= {
"bank",
"bank_id",
"requisition_id",
"invert",
}, "argparser ill defined"
params = [
pfbudget.types.Nordigen(
args["bank"][0],
args["bank_id"][0] if args["bank_id"] else None,
args["requisition_id"][0] if args["requisition_id"] else None,
args["invert"] if args["invert"] else None,
)
]
case pfbudget.Operation.NordigenMod:
assert args.keys() >= {
"bank",
"bank_id",
"requisition_id",
"invert",
"remove",
}, "argparser ill defined"
nargs_1 = ["bank_id", "requisition_id"]
nargs_0 = ["invert"]
param = {"name": args["bank"][0]}
param |= {k: v[0] for k, v in args.items() if k in nargs_1 and args[k]}
param |= {k: v for k, v in args.items() if k in nargs_0}
param |= {k: None for k in args["remove"] if k in nargs_1}
params = [param]
case pfbudget.Operation.NordigenDel:
assert len(args["bank"]) > 0, "argparser ill defined"
params = args["bank"]
case pfbudget.Operation.CategoryAdd:
assert args.keys() >= {"category", "group"}, "argparser ill defined"
params = [

185
pfbudget/cli/runnable.py
View File

@ -5,7 +5,7 @@ import decimal
import re
from pfbudget.common.types import Operation
from pfbudget.db.model import Period
from pfbudget.db.model import AccountType, Period
from pfbudget.input.nordigen import NordigenInput
from pfbudget.db.sqlite import DatabaseClient
import pfbudget.reporting.graph
@ -150,71 +150,24 @@ def argparser() -> argparse.ArgumentParser:
)
p_report.set_defaults(func=report)
"""
Register bank
"""
p_register = subparsers.add_parser(
"register",
description="Register a bank",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
p_register.add_argument(
"--requisition", type=str, nargs=1, help="requisition option help"
)
p_register.add_argument("--invert", action="store_true")
p_register.set_defaults(command=Operation.Register)
# Banks
bank(subparsers.add_parser("bank"))
"""
Unregister bank
"""
p_register = subparsers.add_parser(
"unregister",
description="Unregister a bank",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
p_register.set_defaults(command=Operation.Unregister)
# Nordigen access token
subparsers.add_parser("token").set_defaults(op=Operation.Token)
"""
Nordigen API
"""
p_nordigen_access = subparsers.add_parser(
"token",
description="Get new access token",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_access.set_defaults(command=Operation.Token)
# Nordigen requisition id
requisition = subparsers.add_parser("eua")
requisition.set_defaults(op=Operation.RequisitionId)
requisition.add_argument("name", nargs=1, type=str)
requisition.add_argument("country", nargs=1, type=str)
"""
(Re)new bank requisition ID
"""
p_nordigen_access = subparsers.add_parser(
"renew",
description="(Re)new the Bank requisition ID",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_access.add_argument("name", nargs=1, type=str)
p_nordigen_access.add_argument("country", nargs=1, type=str)
p_nordigen_access.set_defaults(command=Operation.Renew)
"""
Downloading through Nordigen API
"""
p_nordigen_download = subparsers.add_parser(
"download",
description="Downloads transactions using Nordigen API",
parents=[universal, period],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_download.add_argument("--id", nargs="+", type=str)
p_nordigen_download.add_argument("--name", nargs="+", type=str)
p_nordigen_download.add_argument("--all", action="store_true")
p_nordigen_download.set_defaults(command=Operation.Download)
# Download through the Nordigen API
download = subparsers.add_parser("download", parents=[period])
download.set_defaults(op=Operation.Download)
download.add_argument("--id", nargs="+", type=str)
download.add_argument("--name", nargs="+", type=str)
download.add_argument("--all", action="store_true")
# """
# List available banks on Nordigen API
@ -245,11 +198,10 @@ def argparser() -> argparse.ArgumentParser:
# )
# Categories
category_parser = subparsers.add_parser("category", parents=[universal])
category(category_parser, universal)
category(subparsers.add_parser("category"))
# Tag
tags(subparsers.add_parser("tag", parents=[universal]), universal)
tags(subparsers.add_parser("tag"))
# Link
link(subparsers.add_parser("link"))
@ -312,66 +264,107 @@ def report(args):
# input.list(vars(args)["country"][0])
def download(manager, args: dict):
start, end = pfbudget.utils.parse_args_period(args)
manager.parser(NordigenInput(manager, args, start, end))
def category(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
def bank(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add = commands.add_parser("add")
add.set_defaults(op=Operation.BankAdd)
add.add_argument("bank", nargs=1, type=str)
add.add_argument("bic", nargs=1, type=str)
add.add_argument("type", nargs=1, type=str, choices=[e.name for e in AccountType])
rem = commands.add_parser("del")
rem.set_defaults(op=Operation.BankDel)
rem.add_argument("bank", nargs="+", type=str)
mod = commands.add_parser("mod")
mod.set_defaults(op=Operation.BankMod)
mod.add_argument("bank", nargs=1, type=str)
mod.add_argument("--bic", nargs=1, type=str)
mod.add_argument("--type", nargs=1, type=str, choices=[e.name for e in AccountType])
mod.add_argument("--remove", nargs="*", default=[], type=str)
nordigen(commands.add_parser("nordigen"))
def nordigen(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.NordigenAdd)
add.add_argument("bank", nargs=1, type=str)
add.add_argument("--bank_id", nargs=1, type=str)
add.add_argument("--requisition_id", nargs=1, type=str)
add.add_argument("--invert", action="store_true")
rem = commands.add_parser("del")
rem.set_defaults(op=Operation.NordigenDel)
rem.add_argument("bank", nargs="+", type=str)
mod = commands.add_parser("mod")
mod.set_defaults(op=Operation.NordigenMod)
mod.add_argument("bank", nargs=1, type=str)
mod.add_argument("--bank_id", nargs=1, type=str)
mod.add_argument("--requisition_id", nargs=1, type=str)
mod.add_argument("--invert", action="store_true")
mod.add_argument("--remove", nargs="*", default=[], type=str)
def category(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.CategoryAdd)
add.add_argument("category", nargs="+", type=str)
add.add_argument("--group", nargs="?", type=str)
remove = commands.add_parser("remove", parents=[universal])
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.CategoryRemove)
remove.add_argument("category", nargs="+", type=str)
update = commands.add_parser("update", parents=[universal])
update = commands.add_parser("update")
update.set_defaults(op=Operation.CategoryUpdate)
update.add_argument("category", nargs="+", type=str)
update.add_argument("--group", nargs="?", type=str)
schedule = commands.add_parser("schedule", parents=[universal])
schedule = commands.add_parser("schedule")
schedule.set_defaults(op=Operation.CategorySchedule)
schedule.add_argument("category", nargs="+", type=str)
schedule.add_argument("period", nargs=1, choices=[e.value for e in Period])
schedule.add_argument("--frequency", nargs=1, default=[1], type=int)
rule = commands.add_parser("rule", parents=[universal])
category_rule(rule, universal)
rule = commands.add_parser("rule")
category_rule(rule)
group = commands.add_parser("group", parents=[universal])
category_group(group, universal)
group = commands.add_parser("group")
category_group(group)
def category_group(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
def category_group(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add = commands.add_parser("add")
add.set_defaults(op=Operation.GroupAdd)
add.add_argument("group", nargs="+", type=str)
remove = commands.add_parser("remove", parents=[universal])
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.GroupRemove)
remove.add_argument("group", nargs="+", type=str)
def category_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
def category_rule(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add = commands.add_parser("add")
add.set_defaults(op=Operation.RuleAdd)
add.add_argument("category", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove", parents=[universal])
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.RuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify", parents=[universal])
modify = commands.add_parser("modify")
modify.set_defaults(op=Operation.RuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--category", nargs=1, type=str)
@ -379,34 +372,34 @@ def category_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentP
modify.add_argument("--remove", nargs="*", default=[], type=str)
def tags(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
def tags(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add = commands.add_parser("add")
add.set_defaults(op=Operation.TagAdd)
add.add_argument("tag", nargs="+", type=str)
remove = commands.add_parser("remove", parents=[universal])
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.TagRemove)
remove.add_argument("tag", nargs="+", type=str)
rule = commands.add_parser("rule", parents=[universal])
tag_rule(rule, universal)
rule = commands.add_parser("rule")
tag_rule(rule)
def tag_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
def tag_rule(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add", parents=[universal])
add = commands.add_parser("add")
add.set_defaults(op=Operation.TagRuleAdd)
add.add_argument("tag", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove", parents=[universal])
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.TagRuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify", parents=[universal])
modify = commands.add_parser("modify")
modify.set_defaults(op=Operation.TagRuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--tag", nargs=1, type=str)

10
pfbudget/common/types.py
View File

@ -9,10 +9,8 @@ class Operation(Enum):
Parse = auto()
Download = auto()
Categorize = auto()
Register = auto()
Unregister = auto()
Token = auto()
Renew = auto()
RequisitionId = auto()
CategoryAdd = auto()
CategoryUpdate = auto()
CategoryRemove = auto()
@ -29,6 +27,12 @@ class Operation(Enum):
TagRuleModify = auto()
Forge = auto()
Dismantle = auto()
BankAdd = auto()
BankMod = auto()
BankDel = auto()
NordigenAdd = auto()
NordigenMod = auto()
NordigenDel = auto()
class TransactionError(Exception):

53
pfbudget/core/manager.py
View File

@ -1,12 +1,15 @@
from pfbudget.input.input import Input
from pfbudget.input.nordigen import NordigenClient
from pfbudget.input.nordigen import NordigenInput
from pfbudget.input.parsers import parse_data
from pfbudget.db.client import DbClient
from pfbudget.db.model import (
Bank,
Category,
CategoryGroup,
CategoryRule,
CategorySchedule,
Nordigen,
Rule,
Tag,
TagRule,
)
@ -14,7 +17,7 @@ from pfbudget.common.types import Operation
from pfbudget.core.categorizer import Categorizer
from pfbudget.utils import convert
from pfbudget.cli.runnable import download, parse
from pfbudget.cli.runnable import parse
class Manager:
@ -34,9 +37,15 @@ class Manager:
case Operation.Parse:
# TODO this is a monstrosity, remove when possible
parse(self, self.args)
case Operation.Download:
# TODO this is a monstrosity, remove when possible
download(self, self.args)
client = NordigenInput()
client.banks = self.get_banks()
client.start = params[0]
client.end = params[1]
transactions = client.parse()
with self.db.session() as session:
session.add(transactions)
case Operation.Categorize:
with self.db.session() as session:
@ -45,23 +54,29 @@ class Manager:
tags = session.tags()
Categorizer().categorize(uncategorized, categories, tags)
case Operation.Register:
# self._db = DbClient(args["database"])
# self.register(args)
pass
case Operation.Unregister:
# self._db = DbClient(args["database"])
# self.unregister(args)
pass
case Operation.BankMod:
with self.db.session() as session:
session.update(Bank, params)
case Operation.NordigenMod:
with self.db.session() as session:
session.update(Nordigen, params)
case Operation.BankDel:
with self.db.session() as session:
session.remove_by_name(Bank, params)
case Operation.NordigenDel:
with self.db.session() as session:
session.remove_by_name(Nordigen, params)
case Operation.Token:
NordigenClient(self).token()
NordigenInput().token()
case Operation.Renew:
NordigenClient(self).requisition(
self.args["name"], self.args["country"]
)
case Operation.RequisitionId:
NordigenInput().requisition(params[0], params[1])
case Operation.CategoryAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
case Operation.BankAdd | Operation.CategoryAdd | Operation.NordigenAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
with self.db.session() as session:
session.add(params)
@ -94,7 +109,7 @@ class Manager:
case Operation.RuleModify | Operation.TagRuleModify:
assert all(isinstance(param, dict) for param in params)
with self.db.session() as session:
session.updaterules(params)
session.update(Rule, params)
case Operation.GroupAdd:
with self.db.session() as session:

9
pfbudget/db/client.py
View File

@ -114,11 +114,14 @@ class DbClient:
stmt = delete(type).where(type.id.in_(ids))
self.__session.execute(stmt)
def updaterules(self, rules: list[dict]):
self.__session.execute(update(CategoryRule), rules)
def update(self, type, values: list[dict]):
print(type, values)
self.__session.execute(update(type), values)
def remove_links(self, original, links: list):
stmt = delete(Link).where(Link.original == original, Link.link.in_(link for link in links))
stmt = delete(Link).where(
Link.original == original, Link.link.in_(link for link in links)
)
self.__session.execute(stmt)
def uncategorized(self) -> list[Transaction]:

17
pfbudget/db/model.py
View File

@ -59,9 +59,7 @@ class Bank(Base):
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"
)
nordigen: Mapped[Optional[Nordigen]] = relationship(lazy="joined")
bankfk = Annotated[str, mapped_column(Text, ForeignKey(Bank.name))]
@ -79,9 +77,12 @@ class Transaction(Base):
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()
category: Mapped[Optional[TransactionCategory]] = relationship(init=False)
note: Mapped[Optional[Note]] = relationship(init=False)
tags: Mapped[Optional[set[TransactionTag]]] = relationship(init=False)
def __lt__(self, other):
return self.date < other.date
idfk = Annotated[
@ -138,8 +139,6 @@ class Note(Base):
id: Mapped[idfk] = mapped_column(primary_key=True, init=False)
note: Mapped[str]
original: Mapped[Transaction] = relationship(back_populates="note")
class Nordigen(Base):
__tablename__ = "nordigen"
@ -149,8 +148,6 @@ class Nordigen(Base):
requisition_id: Mapped[Optional[str]]
invert: Mapped[Optional[bool]]
bank: Mapped[Bank] = relationship(back_populates="nordigen")
class Tag(Base):
__tablename__ = "tags_available"

16
pfbudget/input/input.py
View File

@ -1,21 +1,9 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from pfbudget.common.types import Transactions
if TYPE_CHECKING:
from pfbudget.core.manager import Manager
from pfbudget.db.model import Transaction
class Input(ABC):
def __init__(self, manager: Manager):
self._manager = manager
@abstractmethod
def parse(self) -> Transactions:
def parse(self) -> list[Transaction]:
return NotImplemented
@property
def manager(self):
return self._manager

66
pfbudget/input/nordigen.py
View File

@ -17,8 +17,8 @@ load_dotenv()
class NordigenInput(Input):
def __init__(self, manager, options: dict = {}, start=date.min, end=date.max):
super().__init__(manager)
def __init__(self):
super().__init__()
self._client = NordigenClient(
secret_key=os.environ.get("SECRET_KEY"),
secret_id=os.environ.get("SECRET_ID"),
@ -28,33 +28,33 @@ class NordigenInput(Input):
# print(options)
if "all" in options and options["all"]:
self.__banks = self.manager.get_banks()
elif "id" in options and options["id"]:
self.__banks = [
self.manager.get_bank_by("nordigen_id", b) for b in options["id"]
]
elif "name" in options and options["name"]:
self.__banks = [
self.manager.get_bank_by("name", b) for b in options["name"]
]
else:
self.__banks = None
# if "all" in options and options["all"]:
# self.__banks = self.manager.get_banks()
# elif "id" in options and options["id"]:
# self.__banks = [
# self.manager.get_bank_by("nordigen_id", b) for b in options["id"]
# ]
# elif "name" in options and options["name"]:
# self.__banks = [
# self.manager.get_bank_by("name", b) for b in options["name"]
# ]
# else:
# self.__banks = None
self.__from = start
self.__to = end
self._start = date.min
self._end = date.max
def parse(self) -> list[Transaction]:
transactions = []
if not self.__banks:
raise NoBankSelected
assert len(self._banks) > 0
for bank in self.__banks:
for bank in self._banks:
print(f"Downloading from {bank}...")
requisition = self.client.requisition.get_requisition_by_id(
bank.nordigen.requisition_id
)
print(requisition)
for acc in requisition["accounts"]:
account = self._client.account_api(acc)
@ -85,10 +85,10 @@ class NordigenInput(Input):
]
transactions.extend(
[t for t in converted if self.__from <= t.date <= self.__to]
[t for t in converted if self._start <= t.date <= self._end]
)
return transactions
return sorted(transactions)
def token(self):
token = self._client.generate_token()
@ -106,6 +106,30 @@ class NordigenInput(Input):
def client(self):
return self._client
@property
def banks(self):
return self._banks
@banks.setter
def banks(self, value):
self._banks = value
@property
def start(self):
return self._start
@start.setter
def start(self, value):
self._start = value
@property
def end(self):
return self._end
@end.setter
def end(self, value):
self._end = value
def __token(self):
if token := os.environ.get("TOKEN"):
return token