167 lines
5.2 KiB
Python
167 lines
5.2 KiB
Python
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.http import HttpResponse
|
|
from django.urls import reverse
|
|
from mixins.views import ObjectDelete, ObjectList, ObjectUpdate
|
|
from rest_framework import status
|
|
from two_factor.views.mixins import OTPRequiredMixin
|
|
|
|
from core.clients.aggregators.nordigen import NordigenClient
|
|
from core.forms import RequisitionForm
|
|
from core.models import Aggregator, Requisition
|
|
from core.util import logs
|
|
from core.views.helpers import synchronize_async_helper
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
class BanksRequisitionUpdate(LoginRequiredMixin, OTPRequiredMixin, ObjectUpdate):
|
|
model = Requisition
|
|
form_class = RequisitionForm
|
|
|
|
page_title = "Update settings for requisition"
|
|
|
|
submit_url_name = "requisition_update"
|
|
submit_url_args = ["type", "aggregator_id", "req_id"]
|
|
|
|
pk_required = False
|
|
|
|
hide_cancel = True
|
|
|
|
def get_object(self, **kwargs):
|
|
aggregator_id = self.kwargs["aggregator_id"]
|
|
req_id = self.kwargs["req_id"]
|
|
|
|
try:
|
|
aggregator = Aggregator.objects.get(
|
|
user=self.request.user, pk=aggregator_id
|
|
)
|
|
except Aggregator.DoesNotExist:
|
|
return HttpResponse(status=status.HTTP_404_NOT_FOUND)
|
|
req, _ = Requisition.objects.get_or_create(
|
|
user=self.request.user,
|
|
aggregator=aggregator,
|
|
requisition_id=req_id,
|
|
)
|
|
return req
|
|
|
|
|
|
class BanksRequisitionDelete(LoginRequiredMixin, OTPRequiredMixin, ObjectDelete):
|
|
model = Requisition
|
|
|
|
|
|
class BanksCurrencies(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
"""
|
|
Get a list of bank accounts with their details.
|
|
"""
|
|
|
|
list_template = "partials/banks-currencies-list.html"
|
|
page_title = "Bank currencies"
|
|
|
|
context_object_name_singular = "currency"
|
|
context_object_name = "currencies"
|
|
|
|
list_url_name = "currencies"
|
|
list_url_args = ["type"]
|
|
|
|
def get_context_data(self):
|
|
context = super().get_context_data()
|
|
self.extra_buttons = [
|
|
{
|
|
"url": reverse("bank_fetch"),
|
|
"action": "refresh",
|
|
"method": "get",
|
|
"label": "Fetch account details",
|
|
"icon": "fa-solid fa-refresh",
|
|
},
|
|
]
|
|
return context
|
|
|
|
def get_queryset(self, **kwargs):
|
|
aggregators = Aggregator.objects.filter(user=self.request.user, enabled=True)
|
|
|
|
account_info = {}
|
|
for agg in aggregators:
|
|
for bank, accounts in agg.account_info.items():
|
|
for account in accounts:
|
|
ident = (bank, account["requisition_id"], account["aggregator_id"])
|
|
if ident not in account_info:
|
|
account_info[ident] = []
|
|
account_info[ident].append(account)
|
|
|
|
return account_info
|
|
|
|
|
|
class BanksBalances(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
"""
|
|
Get the bank balances.
|
|
"""
|
|
|
|
list_template = "partials/banks-balances-list.html"
|
|
page_title = "Bank balances"
|
|
|
|
context_object_name_singular = "balance"
|
|
context_object_name = "balances"
|
|
|
|
list_url_name = "balances"
|
|
list_url_args = ["type"]
|
|
|
|
def get_context_data(self):
|
|
context = super().get_context_data()
|
|
self.extra_buttons = [
|
|
{
|
|
"url": reverse("bank_fetch"),
|
|
"action": "refresh",
|
|
"method": "get",
|
|
"label": "Fetch account details",
|
|
"icon": "fa-solid fa-refresh",
|
|
},
|
|
]
|
|
return context
|
|
|
|
def get_queryset(self, **kwargs):
|
|
aggregators = Aggregator.objects.filter(user=self.request.user, enabled=True)
|
|
account_balances = {}
|
|
for aggregator in aggregators:
|
|
run = synchronize_async_helper(NordigenClient(aggregator))
|
|
balance_map = synchronize_async_helper(run.get_all_balances())
|
|
for k, v in balance_map.items():
|
|
if k not in account_balances:
|
|
account_balances[k] = []
|
|
for item in v:
|
|
account_balances[k].append(item)
|
|
|
|
return account_balances
|
|
|
|
|
|
class BanksTransactions(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
"""
|
|
Get bank transactions.
|
|
"""
|
|
|
|
list_template = "partials/banks-transactions-list.html"
|
|
page_title = "Bank transactions"
|
|
|
|
context_object_name_singular = "transaction"
|
|
context_object_name = "transactions"
|
|
|
|
list_url_name = "transactions"
|
|
list_url_args = ["type", "account_id", "aggregator_id"]
|
|
|
|
def get_queryset(self, **kwargs):
|
|
aggregator_id = self.kwargs.get("aggregator_id")
|
|
account_id = self.kwargs.get("account_id")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(aggregator_id, self.request.user)
|
|
except Aggregator.DoesNotExist:
|
|
context = {
|
|
"message": "Aggregator does not exist",
|
|
"class": "danger",
|
|
}
|
|
return self.render_to_response(context)
|
|
|
|
run = synchronize_async_helper(NordigenClient(aggregator))
|
|
transactions = synchronize_async_helper(
|
|
run.get_transactions(account_id, pending=True)
|
|
)
|
|
return transactions
|