class based view

This commit is contained in:
Corban-Lee Jones 2024-09-23 23:10:48 +01:00
parent 181930d687
commit 9bcb99dd30
2 changed files with 64 additions and 61 deletions

View File

@ -3,10 +3,10 @@
from django.urls import path from django.urls import path
from django.contrib.auth.decorators import login_required from django.contrib.auth.decorators import login_required
from .views import IndexView, guilds from .views import IndexView, GuildsView
urlpatterns = [ urlpatterns = [
path("", login_required(IndexView.as_view()), name="index"), path("", login_required(IndexView.as_view()), name="index"),
path("user-guilds", guilds, name="user-guilds") path("user-guilds", GuildsView.as_view(), name="user-guilds")
] ]

View File

@ -7,7 +7,7 @@ from django.utils import timezone
from asgiref.sync import sync_to_async from asgiref.sync import sync_to_async
from django.shortcuts import redirect from django.shortcuts import redirect
from django.http import JsonResponse, HttpResponse from django.http import JsonResponse, HttpResponse
from django.views.generic import TemplateView from django.views.generic import TemplateView, View
from apps.home.models import r_Server from apps.home.models import r_Server
from apps.authentication.models import DiscordUser, ServerMember from apps.authentication.models import DiscordUser, ServerMember
@ -21,71 +21,74 @@ class IndexView(TemplateView):
template_name = "home/index.html" template_name = "home/index.html"
@sync_to_async class GuildsView(View):
def get_user_access_token(user: DiscordUser) -> str:
return user.access_token
@sync_to_async async def get(self, request, *args, **kwargs):
def is_user_authenticated(user: DiscordUser) -> bool: if not await self.is_user_authenticated(request.user):
return user.is_authenticated return redirect("/oauth2/login")
async def delete_server(server_id: int): access_token = await self.get_user_access_token(request.user)
# member will be auto deleted via CASCADE, so no need to handle that
try: async with httpx.AsyncClient() as client:
server = await r_Server.objects.aget(id=server_id) response = await client.get(
await server.adelete() url=f"{settings.DISCORD_API_URL}/users/@me/guilds",
except r_Server.DoesNotExist: headers={"Authorization": f"Bearer {access_token}"}
pass )
guild_data = response.json()
async def setup_server(user: DiscordUser, data: dict) -> dict | None:
is_owner = data["owner"]
permissions = data["permissions"]
admin_perm = 1 << 3
# Ignore servers where the user isn't an administrator or owner
if not ((int(permissions) & admin_perm) == admin_perm or is_owner):
await delete_server(data["id"])
return
server = await r_Server.objects.aget_or_create(
id=data["id"],
name=data["name"],
icon_hash=data["icon"]
)
await ServerMember.objects.aupdate_or_create(
user=user,
server=server[0],
defaults={
"permissions": permissions,
"is_owner": is_owner
}
)
return data
async def guilds(request): # guilds where the user either administrates or owns
if not await is_user_authenticated(request.user): cleaned_guild_data = []
return redirect("/oauth2/login")
access_token = await get_user_access_token(request.user) for item in guild_data:
cleaned_data = await self.setup_server(request.user, item)
if cleaned_data:
cleaned_guild_data.append(cleaned_data)
async with httpx.AsyncClient() as client: return JsonResponse(cleaned_guild_data, safe=False)
response = await client.get(
url=f"{settings.DISCORD_API_URL}/users/@me/guilds",
headers={"Authorization": f"Bearer {access_token}"} @staticmethod
@sync_to_async
def get_user_access_token(user: DiscordUser) -> str:
return user.access_token
@staticmethod
@sync_to_async
def is_user_authenticated(user: DiscordUser) -> bool:
return user.is_authenticated
async def setup_server(self, user: DiscordUser, data: dict):
is_owner = data["owner"]
permissions = data["permissions"]
admin_perm = 1 << 3
# Ignore servers where the user isn't an administrator or owner
if not ((int(permissions) & admin_perm) == admin_perm or is_owner):
await self.delete_member(user, data["id"])
return
server = await r_Server.objects.aget_or_create(
id=data["id"],
name=data["name"],
icon_hash=data["icon"]
)
await ServerMember.objects.aupdate_or_create(
user=user,
server=server[0],
defaults={
"permissions": permissions,
"is_owner": is_owner
}
) )
guild_data = response.json()
# guilds where the user either administrates or owns return data
cleaned_guild_data = []
for item in guild_data: @staticmethod
cleaned_data = await setup_server(request.user, item) async def delete_member(user: DiscordUser, server_id: int):
if cleaned_data: try:
cleaned_guild_data.append(cleaned_data) member = await ServerMember.objects.aget(user=user, server_id=server_id)
await member.adelete()
except ServerMember.DoesNotExist:
return JsonResponse(guild_data, safe=False) pass