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

Compare commits

base: personal-finance:9500e808de000f354b9a3a98d6a44be357dacf73
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1
..
compare: personal-finance:7fe5b6bd32d0ffdba3934540e7f60c07bc2e7833
Branches Tags
personal-finance:master
personal-finance:0.2
personal-finance:0.1

No commits in common. "9500e808de000f354b9a3a98d6a44be357dacf73" and "7fe5b6bd32d0ffdba3934540e7f60c07bc2e7833" have entirely different histories.
9500e808de ... 7fe5b6bd32

10 changed files with 212 additions and 318 deletions
Show Stats Expand all files Collapse all files

1
pfbudget/__init__.py
View File

@ -7,6 +7,5 @@ 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

100
pfbudget/__main__.py
View File

@ -16,102 +16,6 @@ if __name__ == "__main__":
params = None
match (op):
case pfbudget.Operation.Parse:
assert args.keys() >= {"path", "bank", "creditcard"}
params = [args["path"], args["bank"], args["creditcard"]]
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 = [
@ -153,13 +57,13 @@ if __name__ == "__main__":
params = [
pfbudget.types.CategoryRule(
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,
cat,
)
for cat in args["category"]
]
@ -205,13 +109,13 @@ if __name__ == "__main__":
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,
tag,
)
for tag in args["tag"]
]

222
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 AccountType, Period
from pfbudget.db.model import Period
from pfbudget.input.nordigen import NordigenInput
from pfbudget.db.sqlite import DatabaseClient
import pfbudget.reporting.graph
@ -86,12 +86,20 @@ def argparser() -> argparse.ArgumentParser:
)
p_export.set_defaults(func=lambda args: DatabaseClient(args.database).export())
# Parse from .csv
parse = subparsers.add_parser("parse")
parse.set_defaults(op=Operation.Parse)
parse.add_argument("path", nargs="+", type=str)
parse.add_argument("--bank", nargs=1, type=str)
parse.add_argument("--creditcard", nargs=1, type=str)
"""
Parsing
"""
p_parse = subparsers.add_parser(
"parse",
description="Parses and adds the requested transactions into the selected database",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_parse.add_argument("path", nargs="+", type=str)
p_parse.add_argument("--bank", nargs=1, type=str)
p_parse.add_argument("--creditcard", nargs=1, type=str)
p_parse.add_argument("--category", nargs=1, type=int)
p_parse.set_defaults(command=Operation.Parse)
"""
Categorizing
@ -142,24 +150,71 @@ def argparser() -> argparse.ArgumentParser:
)
p_report.set_defaults(func=report)
# Banks
bank(subparsers.add_parser("bank"))
"""
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)
# Nordigen access token
subparsers.add_parser("token").set_defaults(op=Operation.Token)
"""
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 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)
"""
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)
# 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")
"""
(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)
# """
# List available banks on Nordigen API
@ -190,10 +245,11 @@ def argparser() -> argparse.ArgumentParser:
# )
# Categories
category(subparsers.add_parser("category"))
category_parser = subparsers.add_parser("category", parents=[universal])
category(category_parser, universal)
# Tag
tags(subparsers.add_parser("tag"))
tags(subparsers.add_parser("tag", parents=[universal]), universal)
# Link
link(subparsers.add_parser("link"))
@ -201,6 +257,22 @@ def argparser() -> argparse.ArgumentParser:
return parser
def parse(manager, args):
"""Parses the contents of the path in args to the selected database.
Args:
args (dict): argparse variables
"""
for path in args.path:
if (dir := Path(path)).is_dir():
for file in dir.iterdir():
manager.parse(file, vars(args))
elif Path(path).is_file():
manager.parse(path, vars(args))
else:
raise FileNotFoundError
def graph(args):
"""Plots the transactions over a period of time.
@ -240,142 +312,100 @@ def report(args):
# input.list(vars(args)["country"][0])
def bank(parser: argparse.ArgumentParser):
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):
commands = parser.add_subparsers(required=True)
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 = commands.add_parser("add", parents=[universal])
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")
remove = commands.add_parser("remove", parents=[universal])
remove.set_defaults(op=Operation.CategoryRemove)
remove.add_argument("category", nargs="+", type=str)
update = commands.add_parser("update")
update = commands.add_parser("update", parents=[universal])
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")
schedule = commands.add_parser("schedule", parents=[universal])
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")
category_rule(rule)
rule = commands.add_parser("rule", parents=[universal])
category_rule(rule, universal)
group = commands.add_parser("group")
category_group(group)
group = commands.add_parser("group", parents=[universal])
category_group(group, universal)
def category_group(parser: argparse.ArgumentParser):
def category_group(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add = commands.add_parser("add", parents=[universal])
add.set_defaults(op=Operation.GroupAdd)
add.add_argument("group", nargs="+", type=str)
remove = commands.add_parser("remove")
remove = commands.add_parser("remove", parents=[universal])
remove.set_defaults(op=Operation.GroupRemove)
remove.add_argument("group", nargs="+", type=str)
def category_rule(parser: argparse.ArgumentParser):
def category_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
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")
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")
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)
modify.add_argument("--remove", nargs="*", default=[], type=str)
def tags(parser: argparse.ArgumentParser):
def tags(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
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")
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")
tag_rule(rule)
rule = commands.add_parser("rule", parents=[universal])
tag_rule(rule, universal)
def tag_rule(parser: argparse.ArgumentParser):
def tag_rule(parser: argparse.ArgumentParser, universal: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
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")
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")
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)

10
pfbudget/common/types.py
View File

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

82
pfbudget/core/manager.py
View File

@ -1,17 +1,12 @@
from pathlib import Path
from pfbudget.input.input import Input
from pfbudget.input.nordigen import NordigenInput
from pfbudget.input.nordigen import NordigenClient
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,
)
@ -19,6 +14,8 @@ from pfbudget.common.types import Operation
from pfbudget.core.categorizer import Categorizer
from pfbudget.utils import convert
from pfbudget.cli.runnable import download, parse
class Manager:
def __init__(self, db: str, verbosity: int = 0, args: dict = {}):
@ -34,31 +31,12 @@ class Manager:
match (op):
case Operation.Init:
pass
case Operation.Parse:
# Adapter for the parse_data method. Can be refactored.
args = {"bank": params[1], "creditcard": params[2], "category": None}
transactions = []
for path in params[0]:
if (dir := Path(path)).is_dir():
for file in dir.iterdir():
transactions.extend(self.parse(file, args))
elif Path(path).is_file():
transactions.extend(self.parse(path, args))
else:
raise FileNotFoundError(path)
print(transactions)
if len(transactions) > 0 and input("Commit? (y/n)") == "y":
self.add_transactions(sorted(transactions))
# TODO this is a monstrosity, remove when possible
parse(self, self.args)
case Operation.Download:
client = NordigenInput()
client.banks = self.get_banks()
client.start = params[0]
client.end = params[1]
transactions = client.parse()
self.add_transactions(transactions)
# TODO this is a monstrosity, remove when possible
download(self, self.args)
case Operation.Categorize:
with self.db.session() as session:
@ -67,29 +45,23 @@ class Manager:
tags = session.tags()
Categorizer().categorize(uncategorized, categories, tags)
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.Register:
# self._db = DbClient(args["database"])
# self.register(args)
pass
case Operation.Unregister:
# self._db = DbClient(args["database"])
# self.unregister(args)
pass
case Operation.Token:
NordigenInput().token()
NordigenClient(self).token()
case Operation.RequisitionId:
NordigenInput().requisition(params[0], params[1])
case Operation.Renew:
NordigenClient(self).requisition(
self.args["name"], self.args["country"]
)
case Operation.BankAdd | Operation.CategoryAdd | Operation.NordigenAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
case Operation.CategoryAdd | Operation.RuleAdd | Operation.TagAdd | Operation.TagRuleAdd:
with self.db.session() as session:
session.add(params)
@ -122,7 +94,7 @@ class Manager:
case Operation.RuleModify | Operation.TagRuleModify:
assert all(isinstance(param, dict) for param in params)
with self.db.session() as session:
session.update(Rule, params)
session.updaterules(params)
case Operation.GroupAdd:
with self.db.session() as session:
@ -156,8 +128,14 @@ class Manager:
# client = DatabaseClient(self.__db)
# client.unregister_bank(self.args["bank"][0])
def parse(self, filename: str, args: dict):
return parse_data(filename, args)
def parser(self, parser: Input):
transactions = parser.parse()
print(transactions)
# self.add_transactions(transactions)
# def parse(self, filename: str):
# transactions = parse_data(filename, self.args)
# self.add_transactions(transactions)
# def transactions() -> list[Transaction]:
# pass

9
pfbudget/db/client.py
View File

@ -114,14 +114,11 @@ class DbClient:
stmt = delete(type).where(type.id.in_(ids))
self.__session.execute(stmt)
def update(self, type, values: list[dict]):
print(type, values)
self.__session.execute(update(type), values)
def updaterules(self, rules: list[dict]):
self.__session.execute(update(CategoryRule), rules)
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,7 +59,9 @@ 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(lazy="joined")
nordigen: Mapped[Optional[Nordigen]] = relationship(
back_populates="bank", lazy="joined"
)
bankfk = Annotated[str, mapped_column(Text, ForeignKey(Bank.name))]
@ -77,12 +79,9 @@ class Transaction(Base):
bank: Mapped[bankfk]
amount: Mapped[money]
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
category: Mapped[Optional[TransactionCategory]] = relationship()
note: Mapped[Optional[Note]] = relationship(back_populates="original")
tags: Mapped[Optional[set[TransactionTag]]] = relationship()
idfk = Annotated[
@ -139,6 +138,8 @@ 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"
@ -148,6 +149,8 @@ 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,9 +1,21 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from pfbudget.db.model import Transaction
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) -> list[Transaction]:
def parse(self) -> Transactions:
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):
super().__init__()
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"),
@ -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._start = date.min
self._end = date.max
self.__from = start
self.__to = end
def parse(self) -> list[Transaction]:
transactions = []
assert len(self._banks) > 0
if not self.__banks:
raise NoBankSelected
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._start <= t.date <= self._end]
[t for t in converted if self.__from <= t.date <= self.__to]
)
return sorted(transactions)
return transactions
def token(self):
token = self._client.generate_token()
@ -106,30 +106,6 @@ 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

7
pfbudget/input/parsers.py
View File

@ -4,8 +4,7 @@ from importlib import import_module
import datetime as dt
import yaml
from pfbudget.common.types import NoBankSelected
from pfbudget.db.model import Transaction
from pfbudget.common.types import NoBankSelected, Transaction, Transactions
from pfbudget.utils import utils
Index = namedtuple(
@ -44,7 +43,7 @@ Options = namedtuple(
)
def parse_data(filename: str, args: dict) -> list[Transaction]:
def parse_data(filename: str, args: dict) -> Transactions:
cfg: dict = yaml.safe_load(open("parsers.yaml"))
assert (
"Banks" in cfg
@ -158,7 +157,7 @@ class Parser:
category = line[options.category]
transaction = Transaction(date, text, bank, value, category)
else:
transaction = Transaction(date, text, bank, value)
transaction = Transaction(date, text, bank, value, options.category)
if options.additional_parser:
func(transaction)