pfbudget/__init__.py

@@ -3,5 +3,5 @@ __author__ = "Luís Murta"
 __version__ = "0.1"
 
 from pfbudget.core.categories import categorize_data
+from pfbudget.core.input.parsers import parse_data
 from pfbudget.cli.runnable import run
-from pfbudget.input.parsers import parse_data

pfbudget/cli/runnable.py

@@ -4,15 +4,16 @@ import re
 
 from pfbudget.core.categories import categorize_data
 from pfbudget.core.manager import Manager
-from pfbudget.input.json import JsonParser
+from pfbudget.core.input.json import JsonParser
-from pfbudget.input.nordigen import NordigenInput
 from pfbudget.db.client import DatabaseClient
 import pfbudget.reporting.graph
 import pfbudget.reporting.report
 import pfbudget.utils
 
+from pfbudget.core.input.nordigen import Client
 
-DEFAULT_DB = "data.db"
+
+DEFAULT_DB = "stub.db"
 
 
 class PfBudgetInitialized(Exception):
@@ -163,11 +164,25 @@ 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(
-        "--requisition", type=str, nargs=1, help="requisition option help"
+        "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.add_argument(
+        "--description",
+        type=str,
+        nargs="?",
+        help="description option help"
+    )
     p_register.set_defaults(func=lambda args: manager.register(vars(args)))
 
     """
@@ -179,7 +194,12 @@ 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)))
 
     """
@@ -191,7 +211,7 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
         parents=[help],
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
     )
-    p_nordigen_access.set_defaults(func=lambda args: NordigenInput(manager).token())
+    p_nordigen_access.set_defaults(func=lambda args: Client().token())
 
     """
     Access to Nordigen API
@@ -205,9 +225,7 @@ 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: NordigenInput().requisition(
+        func=lambda args: Client().requisition(args.institution[0], args.country[0])
-            args.institution[0], args.country[0]
-        )
     )
 
     """
@@ -216,16 +234,14 @@ def argparser(manager: Manager) -> argparse.ArgumentParser:
     p_nordigen_download = subparsers.add_parser(
         "download",
         description="Downloads transactions using Nordigen API",
-        parents=[help, period],
+        parents=[help],
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
     )
-    p_nordigen_download.add_argument("--id", nargs="+", type=str)
+    p_nordigen_download.add_argument("id", nargs=1, type=str)
-    p_nordigen_download.add_argument("--name", nargs="+", type=str)
+    p_nordigen_download.set_defaults(func=lambda args: Client().download(args.id[0]))
-    p_nordigen_download.add_argument("--all", action="store_true")
-    p_nordigen_download.set_defaults(func=lambda args: download(manager, args))
 
     """
-    List available banks on Nordigen API
+    List available banks to download from
     """
     p_nordigen_list = subparsers.add_parser(
         "list",
@@ -234,7 +250,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: nordigen_banks(manager, args))
+    p_nordigen_list.set_defaults(func=lambda args: Client().banks(args.country[0]))
 
     """
     Nordigen JSONs
@@ -248,9 +264,7 @@ 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(
+    p_nordigen_json.set_defaults(func=lambda args: manager.parser(JsonParser(vars(args))))
-        func=lambda args: manager.parser(JsonParser(vars(args)))
-    )
 
     return parser
 
@@ -305,16 +319,6 @@ 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 download(manager: Manager, args):
-    start, end = pfbudget.utils.parse_args_period(args)
-    manager.parser(NordigenInput(manager, vars(args), start, end))
-
-
 def run():
     manager = Manager(DEFAULT_DB)
     args = argparser(manager).parse_args()

pfbudget/core/categories.py

@@ -8,8 +8,8 @@ import yaml
 
 
 if TYPE_CHECKING:
-    from pfbudget.common.types import Transaction
     from pfbudget.db.client import DatabaseClient
+    from pfbudget.core.transactions import Transaction
 
 
 Options = namedtuple(

pfbudget/core/input/input.py

@@ -0,0 +1,13 @@
+from abc import ABC, abstractmethod
+
+from pfbudget.core.transactions import Transactions
+
+
+class Input(ABC):
+    @abstractmethod
+    def __init__(self, options: dict):
+        self.options = options
+
+    @abstractmethod
+    def parse(self) -> Transactions:
+        return NotImplemented

pfbudget/core/input/json.py

@@ -1,14 +1,14 @@
 import json
 
-from .input import Input
+from pfbudget.core.input.input import Input
-from pfbudget.common.types import Transactions
+from pfbudget.core.transactions import Transactions
-from pfbudget.utils import convert, parse_decimal
+from pfbudget.utils.converters import convert
+from pfbudget.utils.utils import parse_decimal
 
 
 class JsonParser(Input):
-    def __init__(self, manager, options):
+    def __init__(self, options):
-        super().__init__(manager)
+        super().__init__(options)
-        self.options = options
 
     def parse(self) -> Transactions:
         try:

pfbudget/core/input/nordigen.py

@@ -0,0 +1,84 @@
+from dotenv import load_dotenv
+from nordigen import NordigenClient
+from uuid import uuid4
+import os
+import webbrowser
+
+from .input import Input
+from pfbudget.core.transactions import Transactions
+from pfbudget.utils.converters import convert
+from pfbudget.utils.utils import parse_decimal
+
+load_dotenv()
+
+
+class Client(Input):
+    def __init__(self, options: dict):
+        super().__init__(options)
+        self._client = NordigenClient(
+            secret_key=os.environ.get("SECRET_KEY"),
+            secret_id=os.environ.get("SECRET_ID"),
+        )
+
+        self._client.token = self.__token()
+
+    def parse(self) -> Transactions:
+        requisition = self._client.requisition.get_requisition_by_id(self.options["id"])
+
+        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 t in d["booked"]
+            ]
+
+    def token(self):
+        token = self._client.generate_token()
+        print(f"New access token: {token}")
+        return token
+
+    def requisition(self, institution: str, country: str = "PT"):
+        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):
+        print(self._client.institution.get_institutions(country))
+
+    @property
+    def client(self):
+        return self._client
+
+    def __token(self):
+        if token := os.environ.get("TOKEN"):
+            return token
+        else:
+            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)

pfbudget/core/input/parsers.py

@@ -4,7 +4,7 @@ from importlib import import_module
 import datetime as dt
 import yaml
 
-from pfbudget.common.types import Transaction
+from pfbudget.core.transactions import Transaction
 from pfbudget.utils import utils
 
 Index = namedtuple(

pfbudget/core/manager.py

@@ -1,30 +1,41 @@
-from pfbudget.input.input import Input
+from pfbudget.core.input.input import Input
-from pfbudget.input.parsers import parse_data
+from pfbudget.core.input.parsers import parse_data
-from pfbudget.common.types import Bank, Banks, Transaction, Transactions
+from pfbudget.core.transactions import Transaction
 from pfbudget.db.client import DatabaseClient
-from pfbudget.utils import convert
+from pfbudget.db.schema import Bank
+from pfbudget.utils.converters 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):
-        bank = Bank(args["bank"][0], "", args["requisition"][0], args["invert"])
+        print(args)
-        client = DatabaseClient(self.__db)
+        client = DatabaseClient(self.db)
-        client.register_bank(convert(bank))
+        client.register_bank(
+            Bank(
+                (
+                    args["bank"][0],
+                    args["requisition"][0]
+                    if args["requisition"]
+                    else args["requisition"],
+                    args["invert"],
+                    args["description"],
+                )
+            )
+        )
 
     def unregister(self, args: dict):
-        client = DatabaseClient(self.__db)
+        client = DatabaseClient(self.db)
         client.unregister_bank(args["bank"][0])
 
     def parser(self, parser: Input):
-        transactions = parser.parse()
+        print(parser.parse())
-        self.add_transactions(transactions)
 
     def parse(self, filename: str, args: dict):
         transactions = parse_data(filename, args)
@@ -33,15 +44,6 @@ class Manager:
     def transactions() -> list[Transaction]:
         pass
 
-    def add_transactions(self, transactions: Transactions):
+    def add_transactions(self, transactions: list[Transaction]):
-        client = DatabaseClient(self.__db)
+        converted = convert(transactions)
-        client.insert_transactions([convert(t) for t in transactions])
+        self.__db.insert_transactions(converted)
-
-    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()]

pfbudget/core/transactions.py

@@ -1,7 +1,7 @@
-from dataclasses import dataclass
 from datetime import date
 from decimal import Decimal, InvalidOperation
-from enum import Enum, auto
+
+COMMENT_TOKEN = "#"
 
 
 class TransactionError(Exception):
@@ -104,25 +104,3 @@ class Transaction:
 
 
 Transactions = list[Transaction]
-
-
-class PrimaryKey(Enum):
-    ID = auto()
-    NAME = auto()
-    BIC = auto()
-
-
-@dataclass
-class Bank:
-    name: str
-    bic: str
-    requisition_id: str
-    invert: bool
-    key: PrimaryKey = PrimaryKey.ID
-
-
-Banks = list[Bank]
-
-
-class NoBankSelected(Exception):
-    pass

pfbudget/db/client.py

@@ -7,7 +7,7 @@ import logging.config
 import pathlib
 import sqlite3
 
-from pfbudget.common.types import Transaction
+from pfbudget.core.transactions import Transaction
 import pfbudget.db.schema as Q
 
 
@@ -94,9 +94,9 @@ class DatabaseClient:
         logger.info(f"Adding {transaction} into {self.db}")
         self.__execute(Q.ADD_TRANSACTION, (transaction.to_list(),))
 
-    def insert_transactions(self, transactions: Q.DbTransactions):
+    def insert_transactions(self, transactions: list[list]):
         logger.info(f"Adding {len(transactions)} into {self.db}")
-        self.__executemany(Q.ADD_TRANSACTION, [t.tuple() for t in transactions])
+        self.__executemany(Q.ADD_TRANSACTION, transactions)
 
     def update_category(self, transaction: Transaction):
         logger.info(f"Update {transaction} category")
@@ -190,23 +190,10 @@ class DatabaseClient:
 
     """Banks table methods"""
 
-    def register_bank(self, bank: Q.DbBank):
+    def register_bank(self, bank: Q.Bank):
-        logger.info(f"Registering {bank}")
+        logger.info(f"Registering bank {bank[0]} with req_id={bank[1]}")
-        self.__execute(Q.ADD_BANK, bank.tuple())
+        self.__execute(Q.ADD_BANK, (bank[0], bank[1], bank[2], bank[3]))
 
     def unregister_bank(self, bank: str):
-        logger.info(f"Unregistering {bank}")
+        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 []

pfbudget/db/schema.py

@@ -1,4 +1,3 @@
-from dataclasses import dataclass
 from decimal import Decimal
 
 CREATE_TRANSACTIONS_TABLE = """
@@ -13,27 +12,7 @@ CREATE TABLE IF NOT EXISTS "transactions" (
 );
 """
 
-
+DbTransaction = tuple[str, str | None, str, Decimal, str | None, str | None, str | None]
-@dataclass
-class DbTransaction:
-    date: str
-    description: str
-    bank: str
-    value: Decimal
-    category: str
-    original: str
-    comments: str
-
-    def tuple(self) -> tuple:
-        return (
-            self.date,
-            self.description,
-            self.bank,
-            self.value,
-            self.category,
-        )
-
-
 DbTransactions = list[DbTransaction]
 
 CREATE_BACKUPS_TABLE = """
@@ -46,36 +25,13 @@ CREATE TABLE IF NOT EXISTS backups (
 CREATE_BANKS_TABLE = """
 CREATE TABLE IF NOT EXISTS banks (
     name TEXT NOT NULL PRIMARY KEY,
-    bic TEXT,
+    requisition TEXT,
-    nordigen_id TEXT,
+    invert INTEGER,
-    nordigen_name TEXT,
+    description TEXT
-    requisition_id TEXT,
-    invert INTEGER
 )
 """
 
-
+Bank = tuple[str, str, bool]
-@dataclass
-class DbBank:
-    name: str
-    bic: str
-    nordigen_id: str
-    nordigen_name: str
-    requisition_id: str
-    invert: bool
-
-    def tuple(self):
-        return (
-            self.name,
-            self.bic,
-            self.nordigen_id,
-            self.nordigen_name,
-            self.requisition_id,
-            int(self.invert),
-        )
-
-
-DbBanks = list[DbBank]
 
 ADD_TRANSACTION = """
 INSERT INTO transactions (date, description, bank, value, category) values (?,?,?,?,?)
@@ -138,21 +94,10 @@ ORDER BY date ASC
 """
 
 ADD_BANK = """
-INSERT INTO banks (name, bic, nordigen_id, nordigen_name, requisition_id, invert) values (?,?,?,?,?,?)
+INSERT INTO banks (name, requisition, invert, description) values (?,?,?,?)
 """
 
 DELETE_BANK = """
 DELETE FROM banks
 WHERE name = (?)
 """
-
-SELECT_BANK = """
-SELECT *
-FROM banks
-WHERE {} = (?)
-"""
-
-SELECT_BANKS = """
-SELECT *
-FROM banks
-"""

pfbudget/input/input.py

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

pfbudget/input/nordigen.py

@@ -1,120 +0,0 @@
-from datetime import date
-from time import sleep
-from requests import ReadTimeout
-from dotenv import load_dotenv
-from nordigen import NordigenClient
-from uuid import uuid4
-import os
-import webbrowser
-
-from .input import Input
-from pfbudget.common.types import NoBankSelected, Transactions
-from pfbudget.utils import convert
-
-load_dotenv()
-
-
-class NordigenInput(Input):
-    def __init__(self, manager, options: dict = {}, start=date.min, end=date.max):
-        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()
-
-        # 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.__from = start
-        self.__to = end
-
-    def parse(self) -> Transactions:
-        transactions = []
-        if not self.__banks:
-            raise NoBankSelected
-
-        for bank in self.__banks:
-            print(f"Downloading from {bank}...")
-            requisition = self.client.requisition.get_requisition_by_id(
-                bank.requisition_id
-            )
-
-            for acc in requisition["accounts"]:
-                account = self._client.account_api(acc)
-
-                retries = 0
-                downloaded = {}
-                while retries < 3:
-                    try:
-                        downloaded = account.get_transactions()
-                        break
-                    except ReadTimeout:
-                        retries += 1
-                        print(f"Request #{retries} timed-out, waiting 1s")
-                        sleep(1)
-
-                if not downloaded:
-                    print(f"Couldn't download transactions for {account}")
-                    continue
-
-                converted = [
-                    convert(t, bank.name, bank.invert)
-                    for t in downloaded["transactions"]["booked"]
-                ]
-
-                transactions.extend(
-                    [t for t in converted if self.__from <= t.date <= self.__to]
-                )
-
-        return transactions
-
-    def token(self):
-        token = self._client.generate_token()
-        print(f"New access token: {token}")
-        return token
-
-    def requisition(self, institution: str, country: str = "PT"):
-        link, _ = self.__requisition_id(institution, country)
-        webbrowser.open(link)
-
-    def list(self, country: str):
-        print(self._client.institution.get_institutions(country))
-
-    @property
-    def client(self):
-        return self._client
-
-    def __token(self):
-        if token := os.environ.get("TOKEN"):
-            return token
-        else:
-            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)

pfbudget/utils/__init__.py

@@ -1,2 +0,0 @@
-from .converters import convert
-from .utils import *

pfbudget/utils/converters.py

@@ -1,48 +1,44 @@
 from functools import singledispatch
 
-from pfbudget.common.types import Bank, Transaction, TransactionError
+from pfbudget.core.transactions import Transaction, TransactionError, Transactions
-from pfbudget.db.schema import DbBank, DbTransaction
+from pfbudget.db.schema import DbTransaction, DbTransactions
-from .utils import parse_decimal
 
 
 @singledispatch
 def convert(t):
-    print("No converter as been found")
     pass
 
 
 @convert.register
 def _(t: Transaction) -> DbTransaction:
-    return DbTransaction(t.date, t.description, t.bank, t.value, t.category, t.original, t.additional_comment)
+    return (t.date, t.description, t.bank, t.value, t.category)
 
 
-@convert.register
+def convert_dbtransaction(db) -> Transaction:
-def _(db: DbTransaction) -> Transaction:
     try:
         return Transaction(db)
     except TransactionError:
         print(f"{db} is in the wrong format")
 
 
-@convert.register
+convert.register(type(DbTransaction), convert_dbtransaction)
-def _(db: DbBank, key: str = "") -> Bank:
-    return Bank(db.name, db.bic, db.requisition_id, db.invert, key=key)
 
 
-@convert.register
+def convert_transactions(ts: Transactions) -> DbTransactions:
-def _(bank: Bank, key: str = "") -> DbBank:
-    return DbBank(bank.name, bank.bic, "", "", bank.requisition_id, bank.invert)
-
-
-@convert.register
-def _(json: dict, bank: str, invert: bool) -> Transaction:
-    i = -1 if invert else 1
     try:
-        return Transaction(
+        return [convert(t) for t in ts]
-            json["bookingDate"],
-            json["remittanceInformationUnstructured"],
-            bank,
-            i * parse_decimal(json["transactionAmount"]["amount"]),
-        )
     except TransactionError:
-        print(f"{json} is in the wrong format")
+        print(f"{ts} is in the wrong format")
+
+
+convert.register(type(Transactions), convert_transactions)
+
+
+def convert_dbtransactions(ts: DbTransactions) -> Transactions:
+    try:
+        return [convert(t) for t in ts]
+    except TransactionError:
+        print(f"{ts} is in the wrong format")
+
+
+convert.register(type(DbTransactions), convert_dbtransactions)