pluto/core/clients/aggregators/nordigen.py

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)
        access = response["access"]
        access_expires = response["access_expires"]
        now = timezone.now()
        # Offset now by access_expires seconds
        access_expires = now + timedelta(seconds=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
        )
        if "link" in response:
            return response["link"]
        return False

    async def delete_requisition(self, requisition_id):
        """
        Delete a requisision ID.
        """
        path = f"requisitions/{requisition_id}"

        response = await self.call(
            path, schema="RequisitionDelete", http_method="delete"
        )
        return response