tidy up GuildsView
All checks were successful
Build and Push Docker Image / build (push) Successful in 13s

abstraction
comments
restructure of existing code ...
This commit is contained in:
Corban-Lee Jones 2024-10-11 11:49:26 +01:00
parent 5f0251bd87
commit 2d3f4b6294

View File

@ -6,9 +6,8 @@ import httpx
from django.conf import settings from django.conf import settings
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, HttpResponseNotFound from django.http import JsonResponse, HttpResponseNotFound, HttpResponseNotAllowed
from django.views.generic import TemplateView, View from django.views.generic import TemplateView, View
from django.core.exceptions import PermissionDenied
from apps.home.models import Server, DiscordChannel from apps.home.models import Server, DiscordChannel
from apps.authentication.models import DiscordUser, ServerMember from apps.authentication.models import DiscordUser, ServerMember
@ -34,52 +33,92 @@ def is_user_authenticated(user: DiscordUser) -> bool:
class GuildsView(View): class GuildsView(View):
"""
Fetches the related guilds to the currently authenticated user from Discord.
Will return a filtered list of the results, excluding servers where the
user isn't an administrator or owner.
Valid servers will also be stored in the database for future reference,
along-side the user-server relationship as a member.
"""
async def get(self, request, *args, **kwargs): async def get(self, request, *args, **kwargs):
if not await is_user_authenticated(request.user): if not await is_user_authenticated(request.user):
return redirect("/oauth2/login") return redirect("/oauth2/login")
access_token = await get_user_access_token(request.user) access_token = await get_user_access_token(request.user)
guilds_data, status = await self._get_guilds_data(access_token)
# Send back the error data if status is bad
if status != 200:
return JsonResponse(guilds_data, status=status, safe=False)
cleaned_guilds_data = await self._clean_guilds_data(request.user, guilds_data)
return JsonResponse(cleaned_guilds_data, safe=False)
async def _get_guilds_data(self, access_token: str) -> tuple[list[dict], int]:
"""
Returns the raw guild data and a response status code
from the Discord API.
"""
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
response = await client.get( response = await client.get(
url=f"{settings.DISCORD_API_URL}/users/@me/guilds", url=f"{settings.DISCORD_API_URL}/users/@me/guilds",
headers={"Authorization": f"Bearer {access_token}"} headers={"Authorization": f"Bearer {access_token}"}
) )
status = response.status_code return response.json(), response.status_code
guild_data = response.json()
if status != 200: async def _clean_guilds_data(self, user: DiscordUser, guilds_data: list[dict]) -> list[dict]:
return JsonResponse(guild_data, status=status, safe=False) """
Returns a filtered copy of the given `guilds_data`, without the guilds
where the given `user` is not an administrator or owner of.
# guilds where the user either administrates or owns Also, for each guild, creates/updates an object to represent it, and
cleaned_guild_data = [] another to represent the user/guild relationship as a member.
"""
for item in guild_data: cleaned_guilds_data = []
cleaned_data = await self.setup_server(request.user, item)
if cleaned_data:
cleaned_guild_data.append(cleaned_data)
return JsonResponse(cleaned_guild_data, safe=False) for item in guilds_data:
setup_success = await self._setup_server(user, item)
if setup_success:
cleaned_guilds_data.append(item)
async def setup_server(self, user: DiscordUser, data: dict): return cleaned_guilds_data
is_owner = data["owner"]
permissions = data["permissions"] async def _setup_server(self, user: DiscordUser, item: dict) -> bool:
"""
Create or update a server and user's membership to said server.
Returns `True` if successful, returns `False` if the user isn't an
administrator or owner of the given server.
If `False`, will also attempt to delete any existing member object
linked to the given server.
"""
# Collect some commonly used server data
server_id = item["id"]
is_owner = item["owner"]
permissions = item["permissions"]
admin_perm = 1 << 3 admin_perm = 1 << 3
# Ignore servers where the user isn't an administrator or owner # Skip servers where the user isn't an administrator or owner.
# If an older member object exists, delete that too.
if not ((int(permissions) & admin_perm) == admin_perm or is_owner): if not ((int(permissions) & admin_perm) == admin_perm or is_owner):
await self.delete_member(user, data["id"]) await self._try_delete_member(user, server_id)
return return False
# Create or update an existing server matching the given ID
server = await Server.objects.aupdate_or_create( server = await Server.objects.aupdate_or_create(
id=data["id"], id=server_id,
defaults={ defaults={
"name": data["name"], "name": item["name"],
"icon_hash": data["icon"] "icon_hash": item["icon"]
} }
) )
# Create or update a member object linking the user to the server
await ServerMember.objects.aupdate_or_create( await ServerMember.objects.aupdate_or_create(
user=user, user=user,
server=server[0], server=server[0],
@ -89,10 +128,15 @@ class GuildsView(View):
} }
) )
return data return True
@staticmethod @staticmethod
async def delete_member(user: DiscordUser, server_id: int): async def _try_delete_member(user: DiscordUser, server_id: int):
"""
Attempt to delete any existing server member linked to the given
server id.
"""
try: try:
member = await ServerMember.objects.aget(user=user, server_id=server_id) member = await ServerMember.objects.aget(user=user, server_id=server_id)
await member.adelete() await member.adelete()
@ -101,21 +145,20 @@ class GuildsView(View):
class ChannelsView(View): class ChannelsView(View):
async def get(self, request, *args, **kwargs): async def get(self, request, *args, **kwargs):
if not await is_user_authenticated(request.user): if not await is_user_authenticated(request.user):
return redirect("/oauth2/login") return redirect("/oauth2/login")
guild_id = request.GET.get("guild") guild_id = request.GET.get("guild")
try: try: server = await Server.objects.aget(pk=guild_id)
server = await Server.objects.aget(pk=guild_id) except Server.DoesNotExist: return HttpResponseNotFound("Server not found.")
except Server.DoesNotExist:
return HttpResponseNotFound("Server not found.")
if not ServerMember.objects.filter(server=server, user=request.user).aexists(): if not ServerMember.objects.filter(server=server, user=request.user).aexists():
raise PermissionDenied("You aren't a member of this server.") return HttpResponseNotAllowed("You aren't a member of this server.")
channels_data, status = await self._get_channel_data(guild_id) channels_data, status = await self._get_channel_data(guild_id)
# Send back the error data if status is bad
if status != 200: if status != 200:
return JsonResponse(channels_data, status=status, safe=False) return JsonResponse(channels_data, status=status, safe=False)
@ -159,6 +202,7 @@ class ChannelsView(View):
`data` dictionary, returns the data or `NoneType` if `data['type'] != 0`. `data` dictionary, returns the data or `NoneType` if `data['type'] != 0`.
""" """
# Type 0 = TextChannel, the only one we want
if data.get("type") != 0: if data.get("type") != 0:
return return