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

Adds --all option to download transactions

Browse Source 
There is now the possibility to download the transactions for all banks
in the banks table in the DB.
`NordigenInput` parse method fully functional, and entire chain from
downloading to parsing (simple w/ converter) to writing to DB.
DbTransaction type added __conform__ to simplify writes to DB.
Get bank methods added to both `Manager` and `DatabaseClient`.

Warning: csv parser most likely not working at this point.

Issues #16 and #17
This commit is contained in:
Luís Murta 2022-10-09 18:36:46 +01:00
parent cfcc182f35
commit 9300d42527
Signed by: satprog
GPG Key ID: 169EF1BBD7049F94
7 changed files with 132 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
pfbudget/cli/runnable.py
View File

@ -5,7 +5,7 @@ import re
from pfbudget.core.categories import categorize_data
from pfbudget.core.manager import Manager
from pfbudget.input.json import JsonParser
from pfbudget.input.nordigen import Client
from pfbudget.input.nordigen import NordigenInput
from pfbudget.db.client import DatabaseClient
import pfbudget.reporting.graph
import pfbudget.reporting.report
@ -163,24 +163,13 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
parents=[help],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
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"
"--requisition", type=str, nargs=1, help="requisition option help"
)
p_register.add_argument("--invert", action="store_true")
p_register.add_argument(
"--description",
type=str,
nargs="?",
help="description option help"
"--description", type=str, nargs="?", help="description option help"
)
p_register.set_defaults(func=lambda args: manager.register(vars(args)))
@ -193,12 +182,7 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
parents=[help],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_register.add_argument(
"bank",
type=str,
nargs=1,
help="bank option help"
)
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
p_register.set_defaults(func=lambda args: manager.unregister(vars(args)))
"""
@ -210,7 +194,7 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
parents=[help],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_access.set_defaults(func=lambda args: Client().token())
p_nordigen_access.set_defaults(func=lambda args: NordigenInput(manager).token())
"""
Access to Nordigen API
@ -224,7 +208,9 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
p_nordigen_access.add_argument("institution", nargs=1, type=str)
p_nordigen_access.add_argument("country", nargs=1, type=str)
p_nordigen_access.set_defaults(
func=lambda args: Client().requisition(args.institution[0], args.country[0])
func=lambda args: NordigenInput().requisition(
args.institution[0], args.country[0]
)
)
"""
@ -236,11 +222,15 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
parents=[help],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_download.add_argument("id", nargs=1, type=str)
p_nordigen_download.set_defaults(func=lambda args: Client().download(args.id[0]))
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(
func=lambda args: manager.parser(NordigenInput(manager, vars(args)))
)
"""
List available banks to download from
List available banks on Nordigen API
"""
p_nordigen_list = subparsers.add_parser(
"list",
@ -249,7 +239,7 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_nordigen_list.add_argument("country", nargs=1, type=str)
p_nordigen_list.set_defaults(func=lambda args: Client().banks(args.country[0]))
p_nordigen_list.set_defaults(func=lambda args: nordigen_banks(manager, args))
"""
Nordigen JSONs
@ -263,7 +253,9 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
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))))
p_nordigen_json.set_defaults(
func=lambda args: manager.parser(JsonParser(vars(args)))
)
return parser
@ -318,6 +310,11 @@ def report(args):
pfbudget.reporting.report.detailed(DatabaseClient(args.database), start, end)
def nordigen_banks(manager: Manager, args):
input = NordigenInput(manager)
input.list(vars(args)["country"][0])
def run():
manager = Manager(DEFAULT_DB)
args = argparser(manager).parse_args()

28
pfbudget/core/manager.py
View File

@ -1,21 +1,21 @@
from pfbudget.input.input import Input
from pfbudget.input.parsers import parse_data
from pfbudget.common.types import Bank, Transaction
from pfbudget.common.types import Bank, Banks, Transaction, Transactions
from pfbudget.db.client import DatabaseClient
from pfbudget.utils import convert
class Manager:
def __init__(self, db: str):
self.db = db
self.__db = db
def init(self):
client = DatabaseClient(self.db)
client = DatabaseClient(self.__db)
client.init()
def register(self, args: dict):
print(args)
client = DatabaseClient(self.db)
client = DatabaseClient(self.__db)
client.register_bank(
Bank(
(
@ -30,11 +30,12 @@ class Manager:
)
def unregister(self, args: dict):
client = DatabaseClient(self.db)
client = DatabaseClient(self.__db)
client.unregister_bank(args["bank"][0])
def parser(self, parser: Input):
print(parser.parse())
transactions = parser.parse()
self.add_transactions(transactions)
def parse(self, filename: str, args: dict):
transactions = parse_data(filename, args)
@ -43,6 +44,15 @@ class Manager:
def transactions() -> list[Transaction]:
pass
def add_transactions(self, transactions: list[Transaction]):
converted = convert(transactions)
self.__db.insert_transactions(converted)
def add_transactions(self, transactions: Transactions):
client = DatabaseClient(self.__db)
client.insert_transactions([convert(t) for t in transactions])
def get_bank_by(self, key: str, value: str) -> Bank:
client = DatabaseClient(self.__db)
bank = client.get_bank(key, value)
return convert(bank)
def get_banks(self) -> Banks:
client = DatabaseClient(self.__db)
return [convert(bank) for bank in client.get_banks()]

15
pfbudget/db/client.py
View File

@ -94,7 +94,7 @@ class DatabaseClient:
logger.info(f"Adding {transaction} into {self.db}")
self.__execute(Q.ADD_TRANSACTION, (transaction.to_list(),))
def insert_transactions(self, transactions: list[list]):
def insert_transactions(self, transactions: Q.DbTransactions):
logger.info(f"Adding {len(transactions)} into {self.db}")
self.__executemany(Q.ADD_TRANSACTION, transactions)
@ -197,3 +197,16 @@ class DatabaseClient:
def unregister_bank(self, bank: str):
logger.info(f"Unregistering bank {bank}")
self.__execute(Q.DELETE_BANK, (bank,))
def get_bank(self, key: str, value: str) -> Q.DbBank | None:
logger.info(f"Get bank with {key} = {value}")
bank = self.__execute(Q.SELECT_BANK.format(key), (value, ))
if bank:
return Q.DbBank(*bank[0])
def get_banks(self) -> Q.DbBanks:
logger.info("Get all banks")
banks = self.__execute(Q.SELECT_BANKS)
if banks:
return [Q.DbBank(*bank) for bank in banks]
return []

23
pfbudget/db/schema.py
View File

@ -1,5 +1,6 @@
from dataclasses import dataclass
from decimal import Decimal
import sqlite3
CREATE_TRANSACTIONS_TABLE = """
CREATE TABLE IF NOT EXISTS "transactions" (
@ -24,6 +25,16 @@ class DbTransaction:
original: str
comments: str
def __conform__(self, protocol):
if protocol is sqlite3.PrepareProtocol:
return (
self.date,
self.description,
self.bank,
self.value,
self.category,
)
DbTransactions = list[DbTransaction]
@ -58,7 +69,6 @@ class DbBank:
DbBanks = list[DbBank]
ADD_TRANSACTION = """
INSERT INTO transactions (date, description, bank, value, category) values (?,?,?,?,?)
"""
@ -127,3 +137,14 @@ DELETE_BANK = """
DELETE FROM banks
WHERE name = (?)
"""
SELECT_BANK = """
SELECT *
FROM banks
WHERE {} = (?)
"""
SELECT_BANKS = """
SELECT *
FROM banks
"""

14
pfbudget/input/input.py
View File

@ -1,13 +1,21 @@
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
class Input(ABC):
@abstractmethod
def __init__(self, options: dict):
self.options = options
def __init__(self, manager: Manager):
self._manager = manager
@abstractmethod
def parse(self) -> Transactions:
return NotImplemented
@property
def manager(self):
return self._manager

5
pfbudget/input/json.py
View File

@ -6,8 +6,9 @@ from pfbudget.utils import convert, parse_decimal
class JsonParser(Input):
def __init__(self, options):
super().__init__(options)
def __init__(self, manager, options):
super().__init__(manager)
self.options = options
def parse(self) -> Transactions:
try:

65
pfbudget/input/nordigen.py
View File

@ -5,39 +5,56 @@ import os
import webbrowser
from .input import Input
from pfbudget.common.types import Transactions
from pfbudget.utils import convert, parse_decimal
from pfbudget.common.types import NoBankSelected, Transactions
from pfbudget.utils import convert
load_dotenv()
class Client(Input):
def __init__(self, options: dict):
super().__init__(options)
class NordigenInput(Input):
def __init__(self, manager, options: dict = {}):
super().__init__(manager)
self._client = NordigenClient(
secret_key=os.environ.get("SECRET_KEY"),
secret_id=os.environ.get("SECRET_ID"),
)
self._client.token = self.__token()
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
def parse(self) -> Transactions:
requisition = self._client.requisition.get_requisition_by_id(self.options["id"])
transactions = []
if not self.__banks:
raise NoBankSelected
for acc in requisition["accounts"]:
account = self._client.account_api(acc)
d = account.get_transactions()["transactions"]
return [
convert(
t["bookingDate"],
t["remittanceInformationUnstructured"],
self.options["bank"],
parse_decimal(t["transactionAmount"]["amount"])
if not self.options["invert"]
else -parse_decimal(t["transactionAmount"]["amount"]),
for bank in self.__banks:
requisition = self.client.requisition.get_requisition_by_id(
bank.requisition_id
)
for acc in requisition["accounts"]:
account = self._client.account_api(acc)
d = account.get_transactions()["transactions"]
transactions.extend(
[convert(t, bank.name, bank.invert) for t in d["booked"]]
)
for t in d["booked"]
]
return transactions
def token(self):
token = self._client.generate_token()
@ -48,13 +65,7 @@ class Client(Input):
link, _ = self.__requisition_id(institution, country)
webbrowser.open(link)
def download(self, id: str):
if len(id) > 0:
return self.parse(id)
else:
print("you forgot the req id")
def banks(self, country: str):
def list(self, country: str):
print(self._client.institution.get_institutions(country))
@property