[feat] dkb_raw for standing orders

dkb_robo/dkb_robo.py

@@ -32,15 +32,17 @@ class DKBRobo(object):
     chip_tan = False
     logger = None
     wrapper = None
+    dkb_raw = False
 
-    def __init__(self, dkb_user=None, dkb_password=None, tan_insert=False, legacy_login=False, debug=False, mfa_device=None, chip_tan=False):
+    def __init__(self, dkb_user=None, dkb_password=None, tan_insert=False, legacy_login=False, debug=False, mfa_device=None, chip_tan=False, dkb_raw=False):
         self.dkb_user = dkb_user
         self.dkb_password = dkb_password
         self.chip_tan = chip_tan
         self.tan_insert = tan_insert
         self.legacy_login = legacy_login
         self.logger = logger_setup(debug)
         self.mfa_device = mfa_device
+        self.dkb_raw = dkb_raw
 
     def __enter__(self):
         """ Makes DKBRobo a Context Manager """
@@ -94,7 +96,7 @@ def get_points(self):
     def get_standing_orders(self, uid=None):
         """ get standing orders """
         self.logger.debug('DKBRobo.get_standing_orders()\n')
-        standingorder = StandingOrder(client=self.wrapper.client)
+        standingorder = StandingOrder(client=self.wrapper.client, dkb_raw=self.dkb_raw)
         return standingorder.fetch(uid)
 
     def get_transactions(self, transaction_url, atype, date_from, date_to, transaction_type='booked'):

dkb_robo/standingorder.py

@@ -1,18 +1,71 @@
 """ Module for handling dkb standing orders """
-from typing import Dict, List
+from typing import Dict, List, Optional
+from dataclasses import dataclass, field
 import logging
 import requests
-from dkb_robo.utilities import DKBRoboError
+from dkb_robo.utilities import DKBRoboError, Amount, filter_unexpected_fields
 
 
 logger = logging.getLogger(__name__)
 
 
+@filter_unexpected_fields
+@dataclass
+class CreditorAccount:
+    """ class for a single creditor account """
+    iban: str = None
+    bic: str = None
+    name: str = None
+
+
+@filter_unexpected_fields
+@dataclass
+class DebtorAccount:
+    """ class for a single debitor account """
+    iban: str = None
+    accountId: str = None  # pylint: disable=C0103 # NOSONAR
+
+
+@filter_unexpected_fields
+@dataclass
+class Recurrence:
+    """ class for frequency account """
+    frm: str = None
+    frequency: str = None
+    holidayExecutionStrategy: str = None  # pylint: disable=C0103 # NOSONAR
+    nextExecutionAt: str = None  # pylint: disable=C0103 # NOSONAR
+    until: str = None
+
+
+@filter_unexpected_fields
+@dataclass
+class StandingOrderItem:
+    """ class for a single standing order """
+    amount: Optional[Amount] = None
+    creditor: Optional[CreditorAccount] = None
+    debtor: Optional[DebtorAccount] = None
+    description: str = None
+    messages: List[str] = field(default_factory=list)
+    recurrence: Optional[Recurrence] = None
+    status: str = None
+
+    def __post_init__(self):
+        self.amount = Amount(**self.amount)
+        self.creditor['creditorAccount']['name'] = self.creditor.get('name', {})
+        self.creditor = CreditorAccount(**self.creditor['creditorAccount'])
+        if self.debtor and 'debtorAccount' in self.debtor:
+            self.debtor = DebtorAccount(**self.debtor['debtorAccount'])
+        # rewrite from - field to frm
+        self.recurrence['frm'] = self.recurrence.get('from', None)
+        self.recurrence = Recurrence(**self.recurrence)
+
+
 class StandingOrder:
     """ StandingOrder class """
-    def __init__(self, client: requests.Session, base_url: str = 'https://banking.dkb.de/api'):
+    def __init__(self, client: requests.Session, dkb_raw: bool = False, base_url: str = 'https://banking.dkb.de/api'):
         self.client = client
         self.base_url = base_url
+        self.dkb_raw = dkb_raw
         self.uid = None
 
     def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]:
@@ -23,20 +76,28 @@ def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]:
         if 'data' in full_list:
             for ele in full_list['data']:
 
-                try:
-                    amount = float(ele.get('attributes', {}).get('amount', {}).get('value', None))
-                except Exception as err:
-                    logger.error('amount conversion error: %s', err)
-                    amount = None
-
-                _tmp_dic = {
-                    'amount': amount,
-                    'currencycode': ele.get('attributes', {}).get('amount', {}).get('currencyCode', None),
-                    'purpose': ele.get('attributes', {}).get('description', None),
-                    'recpipient': ele.get('attributes', {}).get('creditor', {}).get('name', None),
-                    'creditoraccount': ele.get('attributes', {}).get('creditor', {}).get('creditorAccount', None),
-                    'interval': ele.get('attributes', {}).get('recurrence', None)}
-                so_list.append(_tmp_dic)
+                standingorder_obj = StandingOrderItem(**ele['attributes'])
+
+                if self.dkb_raw:
+                    so_list.append(standingorder_obj)
+                else:
+                    so_list.append({
+                        'amount': standingorder_obj.amount.value,
+                        'currencycode': standingorder_obj.amount.currencyCode,
+                        'purpose': standingorder_obj.description,
+                        'recipient': standingorder_obj.creditor.name,
+                        'creditoraccount': {
+                            'iban': standingorder_obj.creditor.iban,
+                            'bic': standingorder_obj.creditor.bic
+                        },
+                        'interval': {
+                            'frequency': standingorder_obj.recurrence.frequency,
+                            'from': standingorder_obj.recurrence.frm,
+                            'holidayExecutionStrategy': standingorder_obj.recurrence.holidayExecutionStrategy,
+                            'nextExecutionAt': standingorder_obj.recurrence.nextExecutionAt,
+                            'until': standingorder_obj.recurrence.until
+                        }
+                    })
 
         logger.debug('StandingOrder._filter() ended with: %s entries.', len(so_list))
         return so_list

dkb_robo/utilities.py

@@ -6,11 +6,14 @@
 from string import digits, ascii_letters
 from typing import List, Tuple
 from datetime import datetime, timezone
-from dataclasses import fields
+from dataclasses import dataclass, fields
 import time
 import re
 
 
+logger = logging.getLogger(__name__)
+
+
 def get_dateformat():
     """ get date format """
     return '%d.%m.%Y', '%Y-%m-%d'
@@ -19,6 +22,19 @@ def get_dateformat():
 LEGACY_DATE_FORMAT, API_DATE_FORMAT = get_dateformat()
 JSON_CONTENT_TYPE = 'application/vnd.api+json'
 
+@dataclass
+class Amount:
+    """ Amount data class, roughly based on the JSON API response. """
+    value: float = None
+    currencyCode: str = None
+
+    def __post_init__(self):
+        # convert value to float
+        try:
+            self.value = float(self.value)
+        except Exception as err:
+            logger.error('Account.__post_init: conversion error:  %s', err)
+            self.value = None
 
 class DKBRoboError(Exception):
     """ dkb-robo exception class """

test/test_standingorder.py

@@ -32,6 +32,7 @@ def setUp(self, mock_session):
         self.dir_path = os.path.dirname(os.path.realpath(__file__))
         self.logger = logging.getLogger('dkb_robo')
         self.dkb = StandingOrder(client=mock_session)
+        self.maxDiff = None
 
     @patch('dkb_robo.standingorder.StandingOrder._filter')
     def test_001_fetch(self, mock_filter):
@@ -95,13 +96,14 @@ def test_005__filter(self):
                         }
                     }
                 }]}
-        result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}]
+        result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description', 'recipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01', 'holidayExecutionStrategy': None}}]
         self.assertEqual(result, self.dkb._filter(full_list))
 
     def test_006__filter(self):
         """ test StandingOrder._filter() with list from file """
         so_list = json_load(self.dir_path + '/mocks/so.json')
-        result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description1', 'recpipient': 'name1', 'creditoraccount': {'iban': 'iban1', 'bic': 'bic1'}, 'interval': {'from': '2022-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-01'}}, {'amount': 200.0, 'currencycode': 'EUR', 'purpose': 'description2', 'recpipient': 'name2', 'creditoraccount': {'iban': 'iban2', 'bic': 'bic2'}, 'interval': {'from': '2022-02-01', 'until': '2025-12-02', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-02'}}, {'amount': 300.0, 'currencycode': 'EUR', 'purpose': 'description3', 'recpipient': 'name3', 'creditoraccount': {'iban': 'iban3', 'bic': 'bic3'}, 'interval': {'from': '2022-03-01', 'until': '2025-03-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-03-01'}}]
+        self.dkb.dkb_raw = False
+        result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description1', 'recipient': 'name1', 'creditoraccount': {'iban': 'iban1', 'bic': 'bic1'}, 'interval': {'from': '2022-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-01'}}, {'amount': 200.0, 'currencycode': 'EUR', 'purpose': 'description2', 'recipient': 'name2', 'creditoraccount': {'iban': 'iban2', 'bic': 'bic2'}, 'interval': {'from': '2022-02-01', 'until': '2025-12-02', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-02'}}, {'amount': 300.0, 'currencycode': 'EUR', 'purpose': 'description3', 'recipient': 'name3', 'creditoraccount': {'iban': 'iban3', 'bic': 'bic3'}, 'interval': {'from': '2022-03-01', 'until': '2025-03-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-03-01'}}]
         self.assertEqual(result, self.dkb._filter(so_list))
 
     def test_007__filter(self):
@@ -129,10 +131,10 @@ def test_007__filter(self):
                         }
                     }
                 }]}
-        result = [{'amount': None, 'currencycode': None, 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}]
+        result = [{'amount': None, 'currencycode': None, 'purpose': 'description', 'recipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01', 'holidayExecutionStrategy': None}}]
         with self.assertLogs('dkb_robo', level='INFO') as lcm:
             self.assertEqual(result, self.dkb._filter(full_list))
-        self.assertIn("ERROR:dkb_robo.standingorder:amount conversion error: could not convert string to float: 'aa'", lcm.output)
+        self.assertIn("ERROR:dkb_robo.utilities:Account.__post_init: conversion error:  could not convert string to float: 'aa'", lcm.output)
 
 if __name__ == '__main__':