fints: Ignore unsupported get_cred_card_txs

And fix merging of transaction lists.
master
Jakub Valenta 2 years ago
parent 96e084e1d6
commit e402a9a0ce

@ -8,6 +8,7 @@ from typing import Any, Iterable, Iterator, List, Optional, Sequence, Type
from fints.client import (
FinTS3Client, FinTS3PinTanClient, FinTSOperations, NeedTANResponse,
)
from fints.exceptions import FinTSUnsupportedOperation
from fints.hhd.flicker import terminal_flicker_unix
from fints.models import Account
from fints.utils import minimal_interactive_cli_bootstrap
@ -184,9 +185,6 @@ def get_balance(fc: Type[FinTS3Client], account: Account) -> float:
def fetch_txs(
fc: Type[FinTS3Client], account: Account, *args, **kwargs
) -> Iterator[FinTSTransaction]:
from pprint import pprint as pp
pp(account)
if account.supportedoperations[FinTSOperations.GET_BALANCE]:
logger.info('Fetching balance')
balance = get_balance(fc, account)
@ -199,9 +197,13 @@ def fetch_txs(
FinTSOperations.GET_CREDIT_CARD_TRANSACTIONS
]:
logger.info('Fetching credit card transactions %s %s', args, kwargs)
res = fc.get_credit_card_transactions(
account, account.creditcardnumber, args, kwargs
)
try:
res = fc.get_credit_card_transactions(
account, account.creditcardnumber, args, kwargs
)
except FinTSUnsupportedOperation as e:
logger.error('Failed to fetch credit card transactions: %s', e)
return
else:
raise ValueError(
'Account supports neither GET_TRANSACTIONS nor '
@ -210,15 +212,10 @@ def fetch_txs(
while isinstance(res, NeedTANResponse):
res = ask_for_tan(fc, res)
if not res:
from pprint import pprint as pp
pp(res)
raise FinTSError('No transactions received')
for mt940_tx in res:
logger.info('Reading transaction %s', mt940_tx.data['date'])
tx = FinTSTransaction.from_mt940_tx(mt940_tx, balance)
pp(tx)
yield tx
yield FinTSTransaction.from_mt940_tx(mt940_tx, balance)
def fetch_data(
@ -256,14 +253,14 @@ def fetch_data(
def parse_data(
data: str, backend_id: str, currency: str
) -> Iterator[Transaction]:
txs_list = []
txs_lists = []
for fints_txs_dicts in json.loads(data):
txs = [
FinTSTransaction(**fints_tx_dict).as_transaction(backend_id)
for fints_tx_dict in fints_txs_dicts
]
txs_list.append(txs)
return merge_txs_lists(txs_list)
txs_lists.append(txs)
return merge_txs_lists(txs_lists)
def calc_balance(txs: Iterable[Transaction]) -> Iterator[Transaction]:

@ -32,7 +32,7 @@ def fetch_data(
)
logger.info('Requesting URL "%s".', url)
r = requests.get(url)
assert r.status_code == requests.codes.ok
r.raise_for_status()
return reformat_json(r.text)

@ -10,10 +10,10 @@ def merge_txs_lists(
txs_lists: Sequence[Sequence[Transaction]],
) -> Iterator[Transaction]:
non_empty_txs_lists = [txs for txs in txs_lists if len(txs)]
all_txs = itertools.chain.from_iterable(non_empty_txs_lists)
if len(non_empty_txs_lists) <= 1:
return
yield from all_txs
total_balance = sum(txs[0].balance for txs in non_empty_txs_lists)
all_txs = itertools.chain.from_iterable(non_empty_txs_lists)
all_txs_sorted_by_date = sorted(all_txs, key=operator.attrgetter('date'))
for tx in all_txs_sorted_by_date:
yield dataclasses.replace(tx, balance=total_balance)

Loading…
Cancel
Save