100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
|
from datetime import timedelta
|
||
|
|
|
||
|
|
from django.conf import settings
|
||
|
|
from django.utils import timezone
|
||
|
|
|
||
|
|
from core.clients.base import BaseClient
|
||
|
|
from core.util import logs
|
||
|
|
|
||
|
|
log = logs.get_logger("nordigen")
|
||
|
|
|
||
|
|
|
||
|
|
class NordigenClient(BaseClient):
|
||
|
|
url = "https://ob.nordigen.com/api/v2"
|
||
|
|
|
||
|
|
async def connect(self):
|
||
|
|
now = timezone.now()
|
||
|
|
# Check if access token expires later than now
|
||
|
|
if self.instance.access_token_expires is not None:
|
||
|
|
if self.instance.access_token_expires > now:
|
||
|
|
self.token = self.instance.access_token
|
||
|
|
return
|
||
|
|
await self.get_access_token()
|
||
|
|
|
||
|
|
def method_filter(self, method):
|
||
|
|
new_method = method.replace("/", "_")
|
||
|
|
return new_method
|
||
|
|
|
||
|
|
async def get_access_token(self):
|
||
|
|
"""
|
||
|
|
Get the access token for the Nordigen API.
|
||
|
|
"""
|
||
|
|
log.debug(f"Getting new access token for {self.instance}")
|
||
|
|
data = {
|
||
|
|
"secret_id": self.instance.secret_id,
|
||
|
|
"secret_key": self.instance.secret_key,
|
||
|
|
}
|
||
|
|
|
||
|
|
response = await self.call("token/new", http_method="post", data=data)
|
||
|
|
print("RESPONSE IN GET ACCESS TOKEN", response) #
|
||
|
|
access = response["access"]
|
||
|
|
access_expires = response["access_expires"]
|
||
|
|
print("ACCESS EXPIRES", access_expires)
|
||
|
|
now = timezone.now()
|
||
|
|
# Offset now by access_expires seconds
|
||
|
|
access_expires = now + timedelta(seconds=access_expires)
|
||
|
|
print("ACCESS EXPIRES", access_expires)
|
||
|
|
self.instance.access_token = access
|
||
|
|
self.instance.access_token_expires = access_expires
|
||
|
|
self.instance.save()
|
||
|
|
|
||
|
|
self.token = access
|
||
|
|
|
||
|
|
async def get_requisitions(self):
|
||
|
|
"""
|
||
|
|
Get a list of active accounts.
|
||
|
|
"""
|
||
|
|
response = await self.call("requisitions")
|
||
|
|
return response["results"]
|
||
|
|
|
||
|
|
async def get_countries(self):
|
||
|
|
"""
|
||
|
|
Get a list of countries.
|
||
|
|
"""
|
||
|
|
# This function is a stub.
|
||
|
|
|
||
|
|
return ["GB", "SE"]
|
||
|
|
|
||
|
|
async def get_banks(self, country):
|
||
|
|
"""
|
||
|
|
Get a list of supported banks for a country.
|
||
|
|
:param country: country to query
|
||
|
|
:return: list of institutions
|
||
|
|
:rtype: list
|
||
|
|
"""
|
||
|
|
if not len(country) == 2:
|
||
|
|
return False
|
||
|
|
path = f"institutions/?country={country}"
|
||
|
|
response = await self.call(path, schema="Institutions", append_slash=False)
|
||
|
|
|
||
|
|
return response
|
||
|
|
|
||
|
|
async def build_link(self, institution_id, redirect=None):
|
||
|
|
"""Create a link to access an institution.
|
||
|
|
:param institution_id: ID of the institution
|
||
|
|
"""
|
||
|
|
|
||
|
|
data = {
|
||
|
|
"institution_id": institution_id,
|
||
|
|
"redirect": settings.URL,
|
||
|
|
}
|
||
|
|
if redirect:
|
||
|
|
data["redirect"] = redirect
|
||
|
|
response = await self.call(
|
||
|
|
"requisitions", schema="RequisitionsPost", http_method="post", data=data
|
||
|
|
)
|
||
|
|
print("build_link response", response)
|
||
|
|
if "link" in response:
|
||
|
|
return response["link"]
|
||
|
|
return False
|