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 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()) 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" class AggregatorDelete(LoginRequiredMixin, OTPRequiredMixin, ObjectDelete): model = Aggregator