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

Finish the remaining Nordigen operations

Browse Source 
from the Manager POV and the update on the argparses.
Also clears unnecessary methods from the DB client interface.
Better assert information on the __main__.py
This commit is contained in:
Luís Murta 2023-01-08 19:35:19 +00:00
parent 9b45ee4817
commit 86afa99217
Signed by: satprog
GPG Key ID: 169EF1BBD7049F94
8 changed files with 120 additions and 278 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
pfbudget/__main__.py
View File

@ -17,32 +17,32 @@ if __name__ == "__main__":
params = None
match (op):
case pfbudget.Operation.Parse:
assert args.keys() >= {"path", "bank", "creditcard"}
keys = {"path", "bank", "creditcard"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [args["path"], args["bank"], args["creditcard"]]
case pfbudget.Operation.RequisitionId:
assert args.keys() >= {"name", "country"}, "argparser ill defined"
keys = {"name", "country"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [args["name"][0], args["country"][0]]
case pfbudget.Operation.Download:
assert args.keys() >= {
"id",
"name",
"all",
"interval",
"start",
"end",
"year",
}, "argparser ill defined"
keys = {"all", "banks", "interval", "start", "end", "year", "dry_run"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
start, end = pfbudget.parse_args_period(args)
params = [start, end]
params = [start, end, args["dry_run"]]
if not args["all"]:
params.append(args["banks"])
else:
params.append([])
case pfbudget.Operation.BankAdd:
assert args.keys() >= {
"bank",
"bic",
"type",
}, "argparser ill defined"
keys = {"bank", "bic", "type"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.Bank(
@ -53,12 +53,8 @@ if __name__ == "__main__":
]
case pfbudget.Operation.BankMod:
assert args.keys() >= {
"bank",
"bic",
"type",
"remove",
}, "argparser ill defined"
keys = {"bank", "bic", "type", "remove"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
nargs_1 = ["bic", "type"]
@ -73,12 +69,8 @@ if __name__ == "__main__":
params = args["bank"]
case pfbudget.Operation.NordigenAdd:
assert args.keys() >= {
"bank",
"bank_id",
"requisition_id",
"invert",
}, "argparser ill defined"
keys = {"bank", "bank_id", "requisition_id", "invert"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.Nordigen(
@ -90,13 +82,8 @@ if __name__ == "__main__":
]
case pfbudget.Operation.NordigenMod:
assert args.keys() >= {
"bank",
"bank_id",
"requisition_id",
"invert",
"remove",
}, "argparser ill defined"
keys = {"bank", "bank_id", "requisition_id", "invert", "remove"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
nargs_1 = ["bank_id", "requisition_id"]
nargs_0 = ["invert"]
@ -112,14 +99,24 @@ if __name__ == "__main__":
assert len(args["bank"]) > 0, "argparser ill defined"
params = args["bank"]
case pfbudget.Operation.NordigenCountryBanks:
keys = {"country"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [args["country"][0]]
case pfbudget.Operation.CategoryAdd:
assert args.keys() >= {"category", "group"}, "argparser ill defined"
keys = {"category", "group"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.Category(cat, args["group"]) for cat in args["category"]
]
case pfbudget.Operation.CategoryUpdate:
assert args.keys() >= {"category", "group"}, "argparser ill defined"
keys = {"category", "group"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [pfbudget.types.Category(cat) for cat in args["category"]]
params.append(args["group"])
@ -128,11 +125,8 @@ if __name__ == "__main__":
params = [pfbudget.types.Category(cat) for cat in args["category"]]
case pfbudget.Operation.CategorySchedule:
assert args.keys() >= {
"category",
"period",
"frequency",
}, "argparser ill defined"
keys = {"category", "period", "frequency"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.CategorySchedule(
@ -142,14 +136,8 @@ if __name__ == "__main__":
]
case pfbudget.Operation.RuleAdd:
assert args.keys() >= {
"category",
"date",
"description",
"bank",
"min",
"max",
}, "argparser ill defined"
keys = {"category", "date", "description", "bank", "min", "max"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.CategoryRule(
@ -165,11 +153,13 @@ if __name__ == "__main__":
]
case pfbudget.Operation.RuleRemove | pfbudget.Operation.TagRuleRemove:
assert args.keys() >= {"id"}, "argparser ill defined"
keys = {"id"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = args["id"]
case pfbudget.Operation.RuleModify:
assert args.keys() >= {
keys = {
"id",
"category",
"date",
@ -178,7 +168,8 @@ if __name__ == "__main__":
"min",
"max",
"remove",
}, "argparser ill defined"
}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
nargs_1 = ["category", "date", "description", "regex", "bank", "min", "max"]
params = []
@ -190,18 +181,14 @@ if __name__ == "__main__":
params.append(param)
case pfbudget.Operation.TagAdd:
assert args.keys() >= {"tag"}, "argparser ill defined"
keys = {"tag"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
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"
keys = {"tag", "date", "description", "bank", "min", "max"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.TagRule(
@ -217,16 +204,8 @@ if __name__ == "__main__":
]
case pfbudget.Operation.TagRuleModify:
assert args.keys() >= {
"id",
"tag",
"date",
"description",
"bank",
"min",
"max",
"remove",
}, "argparser ill defined"
keys = {"id", "tag", "date", "description", "bank", "min", "max", "remove"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
nargs_1 = ["tag", "date", "description", "regex", "bank", "min", "max"]
params = []
@ -246,13 +225,17 @@ if __name__ == "__main__":
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"
keys = {"original", "links"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
params = [
pfbudget.types.Link(args["original"][0], link) for link in args["links"]
]
case pfbudget.Operation.Export:
assert args.keys() >= {"interval", "start", "end", "year", "all", "banks", "file"}
keys = {"interval", "start", "end", "year", "all", "banks", "file"}
assert args.keys() >= keys, f"missing {args.keys() - keys}"
start, end = pfbudget.parse_args_period(args)
params = [start, end]
if not args["all"]:

38
pfbudget/cli/runnable.py
View File

@ -151,37 +151,15 @@ def argparser() -> argparse.ArgumentParser:
# 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")
download_banks = download.add_mutually_exclusive_group()
download_banks.add_argument("--all", action="store_true")
download_banks.add_argument("--banks", nargs="+", type=str)
download.add_argument("--dry-run", action="store_true")
# """
# List available banks on Nordigen API
# """
# p_nordigen_list = subparsers.add_parser(
# "list",
# description="Lists banks in {country}",
# parents=[help],
# formatter_class=argparse.ArgumentDefaultsHelpFormatter,
# )
# p_nordigen_list.add_argument("country", nargs=1, type=str)
# p_nordigen_list.set_defaults(func=lambda args: nordigen_banks(manager, args))
# """
# Nordigen JSONs
# """
# p_nordigen_json = subparsers.add_parser(
# "json",
# description="",
# parents=[help],
# formatter_class=argparse.ArgumentDefaultsHelpFormatter,
# )
# p_nordigen_json.add_argument("json", nargs=1, type=str)
# p_nordigen_json.add_argument("bank", nargs=1, type=str)
# p_nordigen_json.add_argument("--invert", action=argparse.BooleanOptionalAction)
# p_nordigen_json.set_defaults(
# func=lambda args: manager.parser(JsonParser(vars(args)))
# )
# List available banks in country C
banks = subparsers.add_parser("banks")
banks.set_defaults(op=Operation.NordigenCountryBanks)
banks.add_argument("country", nargs=1, type=str)
# Categories
category(subparsers.add_parser("category"))

1
pfbudget/common/types.py
View File

@ -33,6 +33,7 @@ class Operation(Enum):
NordigenAdd = auto()
NordigenMod = auto()
NordigenDel = auto()
NordigenCountryBanks = auto()
Export = auto()

36
pfbudget/core/manager.py
View File

@ -1,4 +1,5 @@
from pathlib import Path
import webbrowser
from pfbudget.common.types import Operation
from pfbudget.core.categorizer import Categorizer
@ -8,11 +9,11 @@ from pfbudget.db.model import (
Category,
CategoryGroup,
CategoryRule,
CategorySchedule,
Nordigen,
Rule,
Tag,
TagRule,
Transaction,
)
from pfbudget.input.nordigen import NordigenInput
from pfbudget.input.parsers import parse_data
@ -52,17 +53,27 @@ class Manager:
case Operation.Download:
client = NordigenInput()
client.banks = self.get_banks()
with self.db.session() as session:
if len(params[3]) == 0:
client.banks = session.get(Bank, Bank.nordigen)
else:
client.banks = session.get(Bank, Bank.name, params[3])
session.expunge_all()
client.start = params[0]
client.end = params[1]
transactions = client.parse()
self.add_transactions(transactions)
# dry-run
if not params[2]:
self.add_transactions(transactions)
else:
print(transactions)
case Operation.Categorize:
with self.db.session() as session:
uncategorized = session.uncategorized()
categories = session.categories()
tags = session.tags()
uncategorized = session.get(Transaction, ~Transaction.category)
categories = session.get(Category)
tags = session.get(Tag)
Categorizer().categorize(uncategorized, categories, tags)
case Operation.BankMod:
@ -85,7 +96,13 @@ class Manager:
NordigenInput().token()
case Operation.RequisitionId:
NordigenInput().requisition(params[0], params[1])
link, _ = NordigenInput().requisition(params[0], params[1])
print(f"Opening {link} to request access to {params[0]}")
webbrowser.open(link)
case Operation.NordigenCountryBanks:
banks = NordigenInput().country_banks(params[0])
print(banks)
case Operation.BankAdd | Operation.CategoryAdd | Operation.NordigenAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
with self.db.session() as session:
@ -144,7 +161,7 @@ class Manager:
case Operation.Export:
with self.db.session() as session:
if len(params) < 4:
banks = [bank.name for bank in session.banks()]
banks = [bank.name for bank in session.get(Bank)]
transactions = session.transactions(params[0], params[1], banks)
else:
transactions = session.transactions(
@ -182,9 +199,6 @@ class Manager:
# bank = client.get_bank(key, value)
# return convert(bank)
def get_banks(self):
return self.db.get_nordigen_banks()
@property
def db(self) -> DbClient:
return DbClient(self._db, self._verbosity > 2)

66
pfbudget/db/client.py
View File

@ -2,10 +2,9 @@ from dataclasses import asdict
from datetime import date
from sqlalchemy import create_engine, delete, select, update
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session, joinedload, selectinload
from sqlalchemy.orm import Session
from pfbudget.db.model import (
Bank,
Category,
CategoryGroup,
CategoryRule,
@ -32,39 +31,6 @@ class DbClient:
def __init__(self, url: str, echo=False) -> None:
self._engine = create_engine(url, echo=echo)
def get_transactions(self):
"""¿Non-optimized? get_transactions, will load the entire Transaction"""
with Session(self.engine) as session:
stmt = select(Transaction).options(
joinedload("*"), selectinload(Transaction.tags)
)
return session.scalars(stmt).all()
def get_uncategorized(self):
with Session(self.engine) as session:
stmt = select(Transaction).where(~Transaction.category.has())
return session.scalars(stmt).all()
def get_categorized(self):
with Session(self.engine) as session:
stmt = select(Transaction).where(Transaction.category.has())
return session.scalars(stmt).all()
def insert_transactions(self, input: list[Transaction]):
with Session(self.engine) as session:
session.add_all(input)
session.commit()
def get_banks(self):
with Session(self.engine) as session:
stmt = select(Bank)
return session.scalars(stmt).all()
def get_nordigen_banks(self):
with Session(self.engine) as session:
stmt = select(Bank).where(Bank.nordigen.has())
return session.scalars(stmt).all()
@property
def engine(self):
return self._engine
@ -84,6 +50,20 @@ class DbClient:
def commit(self):
self.__session.commit()
def expunge_all(self):
self.__session.expunge_all()
def get(self, type, column=None, values=None):
if column:
if values:
stmt = select(type).where(column.in_(values))
else:
stmt = select(type).where(column.has())
else:
stmt = select(type)
return self.__session.scalars(stmt).all()
def add(self, rows: list):
self.__session.add_all(rows)
@ -125,10 +105,6 @@ class DbClient:
)
self.__session.execute(stmt)
def uncategorized(self) -> list[Transaction]:
stmt = select(Transaction).where(~Transaction.category.has())
return self.__session.scalars(stmt).all()
def transactions(self, min: date, max: date, banks: list[str]):
stmt = select(Transaction).where(
Transaction.date >= min,
@ -137,17 +113,5 @@ class DbClient:
)
return self.__session.scalars(stmt).all()
def categories(self) -> list[Category]:
stmt = select(Category)
return self.__session.scalars(stmt).all()
def tags(self) -> list[Tag]:
stmt = select(Tag)
return self.__session.scalars(stmt).all()
def banks(self) -> list[Bank]:
stmt = select(Bank)
return self.__session.scalars(stmt).all()
def session(self) -> ClientSession:
return self.ClientSession(self.engine)

30
pfbudget/input/json.py
View File

@ -1,30 +0,0 @@
import json
from .input import Input
from pfbudget.common.types import Transactions
from pfbudget.utils import convert, parse_decimal
class JsonParser(Input):
def __init__(self, manager, options):
super().__init__(manager)
self.options = options
def parse(self) -> Transactions:
try:
with open(self.options["json"][0], "r") as f:
return [
convert(
[
t["bookingDate"],
t["remittanceInformationUnstructured"],
self.options["bank"][0],
parse_decimal(t["transactionAmount"]["amount"])
if not self.options["invert"]
else -parse_decimal(t["transactionAmount"]["amount"]),
],
)
for t in json.load(f)["transactions"]["booked"]
]
except KeyError:
print("No json file defined")

50
pfbudget/input/nordigen.py
View File

@ -6,17 +6,18 @@ from nordigen import NordigenClient
from uuid import uuid4
import json
import os
import webbrowser
from .input import Input
from pfbudget.common.types import NoBankSelected
from pfbudget.db.model import Transaction
from pfbudget.utils import convert
from .input import Input
load_dotenv()
class NordigenInput(Input):
redirect_url = "https://murta.dev"
def __init__(self):
super().__init__()
self._client = NordigenClient(
@ -24,23 +25,7 @@ class NordigenInput(Input):
secret_id=os.environ.get("SECRET_ID"),
)
self.client.token = self.__token()
# 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
self._client.token = self.__token()
self._start = date.min
self._end = date.max
@ -96,11 +81,15 @@ class NordigenInput(Input):
return token
def requisition(self, institution: str, country: str = "PT"):
link, _ = self.__requisition_id(institution, country)
webbrowser.open(link)
id = self._client.institution.get_institution_id_by_name(country, institution)
return self._client.initialize_session(
redirect_uri=self.redirect_url,
institution_id=id,
reference_id=str(uuid4()),
)
def list(self, country: str):
print(self._client.institution.get_institutions(country))
def country_banks(self, country: str):
return self._client.institution.get_institutions(country)
@property
def client(self):
@ -137,16 +126,3 @@ class NordigenInput(Input):
token = self._client.generate_token()
print(f"New access token: {token}")
return token
def __requisition_id(self, i: str, c: str):
id = self._client.institution.get_institution_id_by_name(
country=c, institution=i
)
init = self._client.initialize_session(
redirect_uri="https://murta.dev",
institution_id=id,
reference_id=str(uuid4()),
)
print(f"{i}({c}) link: {init.link} and requisition ID: {init.requisition_id}")
return (init.link, init.requisition_id)

44
pfbudget/utils/converters.py
View File

@ -3,7 +3,6 @@ from functools import singledispatch
from pfbudget.common.types import TransactionError
from pfbudget.db.model import Bank, Transaction
from pfbudget.db.schema import DbBank, DbTransaction
from .utils import parse_decimal
@ -13,49 +12,6 @@ def convert(t):
pass
# @convert.register
# def _(t: Transaction) -> DbTransaction:
# return DbTransaction(
# t.date,
# t.description,
# t.bank,
# t.value,
# t.category,
# t.original,
# t.additional_comment,
# )
# @convert.register
# def _(db: DbTransaction) -> Transaction:
# try:
# return Transaction(db)
# except TransactionError:
# print(f"{db} is in the wrong format")
# @convert.register
# def _(db: DbBank, key: str = "") -> Bank:
# bank = Bank(db.name, db.bic, db.requisition_id, db.invert, db.offset, key=key)
# if not bank.invert:
# bank.invert = False
# if not bank.offset:
# bank.offset = 0
# return bank
# @convert.register
# def _(bank: Bank) -> DbBank:
# bank = DbBank(
# bank.name, bank.bic, "", "", bank.requisition_id, bank.invert, bank.offset
# )
# if not bank.invert:
# bank.invert = False
# if not bank.offset:
# bank.offset = 0
# return bank
@convert.register
def _(json: dict, bank: Bank) -> Transaction:
i = -1 if bank.nordigen.invert else 1