Co-authored-by: Tomek Wójcik <labs@tomekwojcik.pl> Co-committed-by: Tomek Wójcik <labs@tomekwojcik.pl>
168 lines
5.0 KiB
Python
168 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from django.contrib import messages
|
|
import django.db
|
|
from django.http import HttpRequest, HttpResponse, JsonResponse
|
|
from django.shortcuts import redirect, render
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.views.generic import FormView
|
|
from django_htmx.http import trigger_client_event
|
|
|
|
from hotpocket_backend.apps.accounts.decorators import account_required
|
|
from hotpocket_backend.apps.accounts.mixins import AccountRequiredMixin
|
|
from hotpocket_backend.apps.htmx import messages as htmx_messages
|
|
from hotpocket_backend.apps.ui.constants import UIAccessTokenOriginApp
|
|
from hotpocket_backend.apps.ui.dto.accounts import AppsBrowseParams
|
|
from hotpocket_backend.apps.ui.forms.accounts.apps import DeleteForm
|
|
from hotpocket_backend.apps.ui.services import UIAccessTokensService
|
|
from hotpocket_soa.dto.accounts import AccessTokenOut, AccessTokensQuery
|
|
from hotpocket_soa.services import AccessTokensService
|
|
|
|
|
|
class AccessTokenMixin:
|
|
def get_access_token(self) -> AccessTokenOut:
|
|
if hasattr(self, '_access_token') is False:
|
|
setattr(
|
|
self,
|
|
'_access_token',
|
|
UIAccessTokensService().get_or_404(
|
|
account_uuid=self.request.user.pk, # type: ignore[attr-defined]
|
|
pk=self.kwargs['pk'], # type: ignore[attr-defined]
|
|
),
|
|
)
|
|
|
|
return self._access_token # type: ignore[attr-defined]
|
|
|
|
|
|
class DetailView(AccessTokenMixin, AccountRequiredMixin, FormView):
|
|
def get_context_data(self, **kwargs) -> dict:
|
|
result = super().get_context_data(**kwargs)
|
|
|
|
result.update({
|
|
'access_token': self.get_access_token(),
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
class ConfirmationView(DetailView):
|
|
def get_initial(self) -> dict:
|
|
result = super().get_initial()
|
|
|
|
access_token: AccessTokenOut = self.get_access_token()
|
|
|
|
origin_app = access_token.get_origin_app()
|
|
if origin_app is not None:
|
|
origin_app = UIAccessTokenOriginApp[origin_app.value].value
|
|
|
|
result.update({
|
|
'canhazconfirm': 'hai',
|
|
'origin_app': origin_app or access_token.get_origin_app_id(),
|
|
'platform': access_token.meta.get('platform', '-'),
|
|
'version': access_token.meta.get('version', '-'),
|
|
})
|
|
|
|
return result
|
|
|
|
def get_success_url(self) -> str:
|
|
return reverse('ui.accounts.apps.browse')
|
|
|
|
|
|
@account_required
|
|
def index(request: HttpRequest) -> HttpResponse:
|
|
return redirect(reverse('ui.accounts.apps.browse'))
|
|
|
|
|
|
@account_required
|
|
def browse(request: HttpRequest) -> HttpResponse:
|
|
params = AppsBrowseParams.from_request(request=request)
|
|
|
|
access_tokens = AccessTokensService().search(
|
|
query=AccessTokensQuery.model_validate(dict(
|
|
account_uuid=request.user.pk,
|
|
before=params.before,
|
|
)),
|
|
limit=params.limit,
|
|
)
|
|
|
|
before: uuid.UUID | None = None
|
|
if len(access_tokens) == params.limit:
|
|
before = access_tokens[-1].pk
|
|
|
|
next_url: str | None = None
|
|
if before is not None:
|
|
next_url = reverse('ui.accounts.apps.browse', query=[
|
|
('before', before),
|
|
('limit', params.limit),
|
|
])
|
|
|
|
context = {
|
|
'access_tokens': access_tokens,
|
|
'params': params,
|
|
'before': before,
|
|
'next_url': next_url,
|
|
}
|
|
|
|
if request.htmx:
|
|
response = render(
|
|
request,
|
|
'ui/accounts/partials/apps/apps.html',
|
|
context,
|
|
)
|
|
|
|
return trigger_client_event(
|
|
response,
|
|
'HotPocket:BrowseAccountAppsView:updateLoadMoreButton',
|
|
{'next_url': next_url},
|
|
after='swap',
|
|
)
|
|
|
|
return render(
|
|
request,
|
|
'ui/accounts/apps/browse.html',
|
|
context,
|
|
)
|
|
|
|
|
|
class DeleteView(ConfirmationView):
|
|
template_name = 'ui/accounts/apps/delete.html'
|
|
form_class = DeleteForm
|
|
|
|
def form_valid(self, form: DeleteForm) -> HttpResponse:
|
|
with django.db.transaction.atomic():
|
|
result = AccessTokensService().delete(
|
|
access_token=self.get_access_token(),
|
|
)
|
|
|
|
if self.request.htmx:
|
|
response = JsonResponse({
|
|
'status': 'ok',
|
|
'result': result,
|
|
})
|
|
|
|
htmx_messages.add_htmx_message(
|
|
request=self.request,
|
|
response=response,
|
|
level=htmx_messages.SUCCESS,
|
|
message=_('The app auhtorization has been deleted.'),
|
|
)
|
|
|
|
return trigger_client_event(
|
|
response,
|
|
'HotPocket:BrowseAccountAppsView:delete',
|
|
{'pk': self.kwargs['pk']},
|
|
after='swap',
|
|
)
|
|
|
|
messages.add_message(
|
|
self.request,
|
|
messages.SUCCESS,
|
|
_('The app auhtorization has been deleted.'),
|
|
)
|
|
|
|
return super().form_valid(form)
|