import importlib
from typing import Optional
from uuid import UUID
from loguru import logger
from pydantic import AnyUrl
from apiclient.paginators import paginated
from nordigen_cli.apiclient.base import NordigenClientBase
from nordigen_cli.apiclient.endpoints import Endpoints
from nordigen_cli.apiclient.paginator import next_page_by_url
from nordigen_cli.models.model import (
Integration,
IntegrationRetrieve,
EndUserAgreement,
Requisition,
RequisitionRequest,
SpectacularRequisition,
Account,
AccountDetail,
AccountBalance,
AccountTransactions,
)
# inject the requests_debugger module
requests_debugger = importlib.import_module("requests_debugger")
[docs]
class NordigenClient(NordigenClientBase):
# Banks endpoints
[docs]
def list_banks(self, code, /, **kwargs) -> list[Integration]:
response = self.get(self.endpoints.banks.format(code=code), **kwargs)
if isinstance(response, list):
return [Integration.model_validate(bank) for bank in response]
return []
[docs]
def show_bank(self, bank_id, /, **kwargs) -> IntegrationRetrieve:
response = self.get(self.endpoints.bank.format(id=bank_id), **kwargs)
return IntegrationRetrieve.model_validate(response)
# Account endpoints
[docs]
def show_account_details(self, account_id: UUID):
return AccountDetail.model_validate(
self.get(self.endpoints.account_details.format(id=account_id))
)
[docs]
def show_account_balances(self, account_id: UUID):
return AccountBalance.model_validate(
self.get(self.endpoints.account_balances.format(id=account_id))
)
[docs]
def list_account_transactions(self, account_id: UUID) -> AccountTransactions:
response = self.get(self.endpoints.transactions.format(id=account_id))
return AccountTransactions.model_validate(response)
# Agreement endpoints
[docs]
def create_end_user_agreement(self, bank_id, enduser_id, max_historical_days=90):
url = self.endpoints.agreements
data = {
"institution_id": bank_id,
"max_historical_days": max_historical_days,
"access_scope": [
"balances",
"details",
"transactions",
],
}
# //TODO should this be returning json or model?
return self.post(url, data=data).json()
[docs]
def delete_agreement(
self,
agreement_id,
):
logger.debug(f"deleting agreement with id: {agreement_id}")
return self.delete(self.endpoints.agreement.format(id=agreement_id))
[docs]
@paginated(by_url=next_page_by_url)
def list_agreements(
self,
limit: int = 100,
offset: int = 0,
) -> list[EndUserAgreement]:
responses = self.get(self.endpoints.agreements)
results = [
EndUserAgreement.model_validate(result)
for res in responses
for result in res["results"]
]
return results
[docs]
def show_agreement(self, agreement_id: UUID) -> EndUserAgreement:
return EndUserAgreement.model_validate(
self.get(self.endpoints.agreement.format(id=agreement_id))
)
[docs]
def accept_agreement(self, agreement_id, user_agent, ip_address):
url = Endpoints.agreement_accept.format(agreement_id)
data = {"user_agent": user_agent, "ip_address": ip_address}
return self.put(url, data=data).json()
# Requisition endpoints
[docs]
@paginated(by_url=next_page_by_url)
def list_requisitions(
self,
limit: int = 100,
offset: int = 0,
) -> list[Requisition]:
responses = self.get(self.endpoints.requisitions)
results = [
Requisition.model_validate(result)
for res in responses
for result in res["results"]
]
return results
[docs]
def delete_requisition(self, requisition_id: UUID):
return self.delete(self.endpoints.requisition.format(id=requisition_id))
[docs]
def create_requisition(
self,
/,
*,
redirect: str,
institution_id,
agreement: UUID = None,
reference: str = None,
user_language="EN",
ssn: Optional[str],
account_selection: Optional[bool] = False,
redirect_immediate: Optional[bool] = False,
) -> SpectacularRequisition:
url = self.endpoints.requisitions
logger.trace(f"creating requisition with reference: {reference}")
logger.trace(f"ssn is: {ssn}")
logger.trace(f"user_language is: {user_language}")
request = RequisitionRequest(
redirect=AnyUrl(redirect),
institution_id=institution_id,
agreement=agreement,
reference=reference,
user_language=user_language,
ssn=ssn,
account_selection=account_selection,
redirect_immediate=redirect_immediate,
)
return SpectacularRequisition.model_validate(
self.post(
url,
data=request.model_dump(
mode="json",
exclude_none=True,
),
)
)
[docs]
def show_requisition(self, requisition_id):
return Requisition.model_validate(
self.get(self.endpoints.requisition.format(id=requisition_id))
)