Allow configuring aggregator connections

This commit is contained in:
Mark Veidemanis 2023-03-07 16:59:39 +00:00
parent d094481583
commit c702e6ecea
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
7 changed files with 242 additions and 2 deletions

View File

@ -20,7 +20,7 @@ from django.contrib.auth.views import LogoutView
from django.urls import include, path
from two_factor.urls import urlpatterns as tf_urls
from core.views import base, notifications
from core.views import aggregators, base, notifications
# from core.views.stripe_callbacks import Callback
@ -32,9 +32,31 @@ urlpatterns = [
path("", include(tf_urls)),
path("accounts/signup/", base.Signup.as_view(), name="signup"),
path("accounts/logout/", LogoutView.as_view(), name="logout"),
# Notifications
path(
"notifications/<str:type>/update/",
notifications.NotificationsUpdate.as_view(),
name="notifications_update",
),
# Aggregators
path(
"aggs/<str:type>/",
aggregators.AggregatorList.as_view(),
name="aggregators",
),
path(
"aggs/<str:type>/create/",
aggregators.AggregatorCreate.as_view(),
name="aggregator_create",
),
path(
"aggs/<str:type>/update/<str:pk>/",
aggregators.AggregatorUpdate.as_view(),
name="aggregator_update",
),
path(
"aggs/<str:type>/delete/<str:pk>/",
aggregators.AggregatorDelete.as_view(),
name="aggregator_delete",
),
] + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)

View File

@ -4,7 +4,7 @@ from django.core.exceptions import FieldDoesNotExist
from django.forms import ModelForm
from mixins.restrictions import RestrictedFormMixin
from .models import NotificationSettings, User
from .models import Aggregator, NotificationSettings, User
# flake8: noqa: E501
@ -48,3 +48,26 @@ class NotificationSettingsForm(RestrictedFormMixin, ModelForm):
"ntfy_topic": "The topic to send notifications to.",
"ntfy_url": "Custom NTFY server. Leave blank to use the default server.",
}
class AggregatorForm(RestrictedFormMixin, ModelForm):
def __init__(self, *args, **kwargs):
super(AggregatorForm, self).__init__(*args, **kwargs)
self.fields["secret_id"].label = "Secret ID"
class Meta:
model = Aggregator
fields = (
"name",
"service",
"secret_id",
"secret_key",
"poll_interval",
)
help_texts = {
"name": "The name of the aggregator connection.",
"service": "The aggregator service to use.",
"secret_id": "The secret ID for the aggregator service.",
"secret_key": "The secret key for the aggregator service.",
"poll_interval": "The interval in seconds to poll the aggregator service.",
}

View File

@ -0,0 +1,35 @@
# Generated by Django 4.1.7 on 2023-03-07 16:54
import uuid
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='notificationsettings',
name='user',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL),
),
migrations.CreateModel(
name='Aggregator',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=255)),
('service', models.CharField(choices=[('nordigen', 'Nordigen')], max_length=255)),
('secret_id', models.CharField(blank=True, max_length=1024, null=True)),
('secret_key', models.CharField(blank=True, max_length=1024, null=True)),
('access_token', models.CharField(blank=True, max_length=1024, null=True)),
('poll_interval', models.IntegerField(default=10)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
),
]

View File

@ -31,7 +31,15 @@ class NotificationSettings(models.Model):
class Aggregator(models.Model):
"""
A connection to an API aggregator to pull transactions from bank accounts.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(User, on_delete=models.CASCADE)
name = models.CharField(max_length=255)
service = models.CharField(max_length=255, choices=SERVICE_CHOICES)
secret_id = models.CharField(max_length=1024, null=True, blank=True)
secret_key = models.CharField(max_length=1024, null=True, blank=True)
access_token = models.CharField(max_length=1024, null=True, blank=True)
poll_interval = models.IntegerField(default=10)

View File

@ -219,6 +219,20 @@
Home
</a>
{% if user.is_authenticated %}
<div class="navbar-item has-dropdown is-hoverable">
<a class="navbar-link">
Setup
</a>
<div class="navbar-dropdown">
<a class="navbar-item" href="{% url 'aggregators' type='page' %}">
Bank Aggregators
</a>
<a class="navbar-item" href="#">
Platform Connections
</a>
</div>
</div>
<div class="navbar-item has-dropdown is-hoverable">
<a class="navbar-link">
Account

View File

@ -0,0 +1,94 @@
{% load cache %}
{% load cachalot cache %}
{% get_last_invalidation 'core.Hook' as last %}
{% include 'mixins/partials/notify.html' %}
{# cache 600 objects_hooks request.user.id object_list type last #}
<table
class="table is-fullwidth is-hoverable"
hx-target="#{{ context_object_name }}-table"
id="{{ context_object_name }}-table"
hx-swap="outerHTML"
hx-trigger="{{ context_object_name_singular }}Event from:body"
hx-get="{{ list_url }}">
<thead>
<th>id</th>
<th>user</th>
<th>name</th>
<th>service</th>
<th>actions</th>
</thead>
{% for item in object_list %}
<tr>
<td>
<a
class="has-text-grey"
onclick="window.prompt('Copy to clipboard: Ctrl+C, Enter', '{{ item.id }}/');">
<span class="icon" data-tooltip="Copy to clipboard">
<i class="fa-solid fa-copy" aria-hidden="true"></i>
</span>
</a>
</td>
<td>{{ item.user }}</td>
<td>{{ item.name }}</td>
<td>{{ item.get_service_display }}</td>
<td>
<div class="buttons">
<button
hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
hx-get="{% url 'aggregator_update' type=type pk=item.id %}"
hx-trigger="click"
hx-target="#{{ type }}s-here"
hx-swap="innerHTML"
class="button">
<span class="icon-text">
<span class="icon">
<i class="fa-solid fa-pencil"></i>
</span>
</span>
</button>
<button
hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
hx-delete="{% url 'aggregator_delete' type=type pk=item.id %}"
hx-trigger="click"
hx-target="#modals-here"
hx-swap="innerHTML"
hx-confirm="Are you sure you wish to delete {{ item.name }}?"
class="button">
<span class="icon-text">
<span class="icon">
<i class="fa-solid fa-xmark"></i>
</span>
</span>
</button>
{% if type == 'page' %}
<a href="#"><button
class="button">
<span class="icon-text">
<span class="icon">
<i class="fa-solid fa-eye"></i>
</span>
</span>
</button>
</a>
{% else %}
<button
hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
hx-get="#"
hx-trigger="click"
hx-target="#{{ type }}s-here"
hx-swap="innerHTML"
class="button">
<span class="icon-text">
<span class="icon">
<i class="fa-solid fa-eye"></i>
</span>
</span>
</button>
{% endif %}
</div>
</td>
</tr>
{% endfor %}
</table>
{# endcache #}

44
core/views/aggregators.py Normal file
View File

@ -0,0 +1,44 @@
from django.contrib.auth.mixins import LoginRequiredMixin
from mixins.views import (
ObjectCreate,
ObjectDelete,
ObjectList,
ObjectRead,
ObjectUpdate,
)
from two_factor.views.mixins import OTPRequiredMixin
from core.forms import AggregatorForm
from core.models import Aggregator
from core.util import logs
log = logs.get_logger(__name__)
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