You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
282 lines
8.9 KiB
Python
282 lines
8.9 KiB
Python
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.http import HttpResponse
|
|
from django.shortcuts import render
|
|
from django.urls import reverse
|
|
from django.views import View
|
|
from mixins.views import (
|
|
ObjectCreate,
|
|
ObjectDelete,
|
|
ObjectList,
|
|
ObjectRead,
|
|
ObjectUpdate,
|
|
)
|
|
from two_factor.views.mixins import OTPRequiredMixin
|
|
|
|
from core.clients.aggregators.nordigen import NordigenClient
|
|
from core.forms import AggregatorForm
|
|
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 RequestBankFetch(LoginRequiredMixin, OTPRequiredMixin, View):
|
|
template_name = "mixins/partials/notify.html"
|
|
|
|
def get(self, request, pk=None):
|
|
if pk:
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, self.request.user)
|
|
aggregators = [aggregator]
|
|
|
|
except Aggregator.DoesNotExist:
|
|
message = "Aggregator does not exist"
|
|
context = {
|
|
"message": message,
|
|
"class": "danger",
|
|
}
|
|
return self.render_to_response(context)
|
|
else:
|
|
aggregators = Aggregator.objects.filter(user=self.request.user)
|
|
|
|
for agg in aggregators:
|
|
agg.fetch_accounts = True
|
|
agg.save()
|
|
|
|
context = {"class": "success", "message": "Fetch requested"}
|
|
return render(request, self.template_name, context)
|
|
|
|
|
|
class ReqsList(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
list_template = "partials/aggregator-info.html"
|
|
page_title = "Aggregator info"
|
|
|
|
context_object_name_singular = "requisition"
|
|
context_object_name = "requisitions"
|
|
|
|
list_url_name = "reqs"
|
|
list_url_args = ["type", "pk"]
|
|
|
|
submit_url_name = "aggregator_countries"
|
|
submit_url_args = ["type", "pk"]
|
|
|
|
def get_context_data(self):
|
|
context = super().get_context_data()
|
|
pk = self.kwargs.get("pk")
|
|
context["pk"] = pk
|
|
self.extra_buttons = [
|
|
{
|
|
"url": reverse("bank_fetch", kwargs={"pk": pk}),
|
|
"action": "refresh",
|
|
"method": "get",
|
|
"label": "Fetch account details",
|
|
"icon": "fa-solid fa-refresh",
|
|
},
|
|
]
|
|
return context
|
|
|
|
def get_queryset(self, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, self.request.user)
|
|
|
|
except Aggregator.DoesNotExist:
|
|
context = {
|
|
"message": "Aggregator does not exist",
|
|
"class": "danger",
|
|
}
|
|
return self.render_to_response(context)
|
|
|
|
self.page_title = (
|
|
f"Requisitions for {aggregator.name} ({aggregator.get_service_display()})"
|
|
)
|
|
self.page_subtitle = f"Stored account details: {len(aggregator.currencies)}"
|
|
|
|
run = synchronize_async_helper(NordigenClient(aggregator))
|
|
reqs = synchronize_async_helper(run.get_requisitions())
|
|
for req in reqs:
|
|
# Add in Requisition object
|
|
requisition_id = req["id"]
|
|
requisition = Requisition.objects.filter(
|
|
user=self.request.user,
|
|
aggregator=aggregator,
|
|
requisition_id=requisition_id,
|
|
).first()
|
|
if requisition:
|
|
req["requisition"] = requisition
|
|
return reqs
|
|
|
|
|
|
class AggregatorCountriesList(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
list_template = "partials/aggregator-countries.html"
|
|
page_title = "List of countries"
|
|
|
|
list_url_name = "aggregator_countries"
|
|
list_url_args = ["type", "pk"]
|
|
|
|
context_object_name_singular = "country"
|
|
context_object_name = "countries"
|
|
|
|
def get_context_data(self):
|
|
context = super().get_context_data()
|
|
context["pk"] = self.kwargs.get("pk")
|
|
return context
|
|
|
|
def get_queryset(self, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, self.request.user)
|
|
|
|
except Aggregator.DoesNotExist:
|
|
context = {
|
|
"message": "Aggregator does not exist",
|
|
"class": "danger",
|
|
}
|
|
return self.render_to_response(context)
|
|
|
|
self.page_title = (
|
|
f"Countries for {aggregator.name} ({aggregator.get_service_display()})"
|
|
)
|
|
run = synchronize_async_helper(NordigenClient(aggregator))
|
|
countries = synchronize_async_helper(run.get_countries())
|
|
self.extra_args = {"pk": pk}
|
|
return countries
|
|
|
|
|
|
class AggregatorCountryBanksList(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
list_template = "partials/aggregator-country-banks.html"
|
|
page_title = "List of banks"
|
|
|
|
list_url_name = "aggregator_country_banks"
|
|
list_url_args = ["type", "pk", "country"]
|
|
|
|
context_object_name_singular = "bank"
|
|
context_object_name = "banks"
|
|
|
|
def get_context_data(self):
|
|
context = super().get_context_data()
|
|
context["pk"] = self.kwargs.get("pk")
|
|
context["country"] = self.kwargs.get("country")
|
|
return context
|
|
|
|
def get_queryset(self, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
country = kwargs.get("country")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, self.request.user)
|
|
|
|
except Aggregator.DoesNotExist:
|
|
context = {
|
|
"message": "Aggregator does not exist",
|
|
"class": "danger",
|
|
}
|
|
return self.render_to_response(context)
|
|
|
|
self.page_title = (
|
|
f"Banks for {aggregator.name} in {country} "
|
|
f"({aggregator.get_service_display()})"
|
|
)
|
|
run = synchronize_async_helper(NordigenClient(aggregator))
|
|
banks = synchronize_async_helper(run.get_banks(country))
|
|
return banks
|
|
|
|
|
|
class AggregatorLinkBank(LoginRequiredMixin, OTPRequiredMixin, View):
|
|
def get(self, request, *args, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
bank = kwargs.get("bank")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, 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))
|
|
auth_url = synchronize_async_helper(run.build_link(bank))
|
|
|
|
# Create a blank response
|
|
response = HttpResponse()
|
|
response["HX-Redirect"] = auth_url
|
|
return response
|
|
|
|
|
|
class ReqDelete(LoginRequiredMixin, OTPRequiredMixin, View):
|
|
def delete(self, request, *args, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
req_id = kwargs.get("req_id")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, 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))
|
|
synchronize_async_helper(run.delete_requisition(req_id))
|
|
response = HttpResponse(status=204)
|
|
response["HX-Trigger"] = "requisitionEvent"
|
|
return response
|
|
|
|
|
|
class ReqInfo(LoginRequiredMixin, OTPRequiredMixin, ObjectRead):
|
|
context_object_name_singular = "requisition"
|
|
context_object_name = "requisitions"
|
|
detail_template = "partials/aggregator-req-info.html"
|
|
|
|
def get_object(self, **kwargs):
|
|
pk = kwargs.get("pk")
|
|
req_id = kwargs.get("req_id")
|
|
try:
|
|
aggregator = Aggregator.get_by_id(pk, 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))
|
|
req = synchronize_async_helper(run.get_all_account_info(req_id))
|
|
self.extra_context = {"pretty": list(req.keys())}
|
|
return req
|
|
|
|
|
|
class AggregatorList(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
|
|
list_template = "partials/aggregator-list.html"
|
|
model = Aggregator
|
|
page_title = "List of aggregator connections"
|
|
|
|
list_url_name = "aggregators"
|
|
list_url_args = ["type"]
|
|
|
|
submit_url_name = "aggregator_create"
|
|
|
|
|
|
class AggregatorCreate(LoginRequiredMixin, OTPRequiredMixin, ObjectCreate):
|
|
model = Aggregator
|
|
form_class = AggregatorForm
|
|
|
|
submit_url_name = "aggregator_create"
|
|
|
|
|
|
class AggregatorUpdate(LoginRequiredMixin, OTPRequiredMixin, ObjectUpdate):
|
|
model = Aggregator
|
|
form_class = AggregatorForm
|
|
|
|
submit_url_name = "aggregator_update"
|
|
|
|
def post_save(self, obj):
|
|
obj.access_token = None
|
|
obj.access_token_expires = None
|
|
obj.save()
|
|
|
|
|
|
class AggregatorDelete(LoginRequiredMixin, OTPRequiredMixin, ObjectDelete):
|
|
model = Aggregator
|