# 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.token = None 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: self.token = parsed["access"] self.log.info("Refreshed access token - Nordigen") 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 {self.token}"} 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