handler/sinks/nordigen.py

@@ -0,0 +1,66 @@
+# Twisted/Klein imports
+from twisted.logger import Logger
+
+# Other library imports
+import requests
+from json import dumps
+from simplejson.errors import JSONDecodeError
+
+# Project imports
+from settings import settings
+
+
+class Nordigen(object):
+    """
+    Class to manage calls to Open Banking APIs through Nordigen.
+    """
+
+    def __init__(self):
+        self.log = Logger("nordigen")
+        # self.get_access_token()
+
+    def get_access_token(self):
+        """
+        Get an access token.
+        :return: True or False
+        :rtype: bool
+        """
+        headers = {"accept": "application/json", "Content-Type": "application/json"}
+        data = {
+            "secret_id": settings.Nordigen.ID,
+            "secret_key": settings.Nordigen.Key,
+        }
+        path = f"{settings.Nordigen.Base}/token/new/"
+        r = requests.post(path, headers=headers, data=dumps(data))
+        try:
+            parsed = r.json()
+        except JSONDecodeError:
+            self.log.error("Error parsing access token response: {content}", content=r.content)
+            return False
+        if "access" in parsed:
+            settings.Nordigen.Access = parsed["access"]
+            settings.write()
+
+        print(dumps(self.get_institutions(""), indent=2))
+
+    def get_institutions(self, country, filter_name=None):
+        """
+        Get a list of supported institutions.
+        """
+        if not len(country) == 2:
+            return False
+        headers = {"accept": "application/json", "Authorization": f"Bearer {settings.Nordigen.Access}"}
+        path = f"{settings.Nordigen.Base}/institutions/?country={country}"
+        r = requests.get(path, headers=headers)
+        try:
+            parsed = r.json()
+        except JSONDecodeError:
+            self.log.error("Error parsing institutions response: {content}", content=r.content)
+            return False
+        new_list = []
+        if filter_name:
+            for i in parsed:
+                if filter_name in i["name"]:
+                    new_list.append(i)
+            return new_list
+        return parsed