334 lines
9.8 KiB
Python
334 lines
9.8 KiB
Python
import graphene
|
|
from django.contrib.auth import update_session_auth_hash
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.db import IntegrityError
|
|
from django.db.models import Q
|
|
from graphene import relay
|
|
from graphql_relay import from_global_id
|
|
from api.types import FailureNode
|
|
|
|
from api.utils import get_object
|
|
from core.logger import get_logger
|
|
from users.inputs import PasswordUpdateInput
|
|
from users.models import SchoolClass, SchoolClassMember, Team
|
|
from users.schema import (
|
|
CreateSchoolClassResult,
|
|
CreateTeamResult,
|
|
FieldError,
|
|
SchoolClassNode,
|
|
TeamNode,
|
|
UpdateError,
|
|
DuplicateName,
|
|
)
|
|
from users.serializers import AvatarUrlSerializer, PasswordSerialzer
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
DuplicateFailure = DuplicateName(reason="Dieser Name wird bereits verwendet.")
|
|
|
|
|
|
class CodeNotFoundException(Exception):
|
|
pass
|
|
|
|
|
|
class TeacherOnlyMutation(relay.ClientIDMutation):
|
|
class Meta:
|
|
abstract = True
|
|
|
|
@classmethod
|
|
def mutate(cls, root, info, input):
|
|
user = info.context.user
|
|
if "users.can_manage_school_class_content" not in user.get_role_permissions():
|
|
raise PermissionError("Permission denied")
|
|
return super().mutate(root, info, input)
|
|
|
|
|
|
class UpdatePassword(relay.ClientIDMutation):
|
|
class Input:
|
|
password_input = graphene.Argument(PasswordUpdateInput)
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(UpdateError) # todo: change for consistency
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
password_data = kwargs.get("password_input")
|
|
|
|
serializer = PasswordSerialzer(data=password_data, context=user)
|
|
|
|
if serializer.is_valid():
|
|
user.set_password(password_data["new_password"])
|
|
user.save()
|
|
update_session_auth_hash(info.context, user)
|
|
return cls(success=True)
|
|
|
|
errors = []
|
|
for key, value in serializer.errors.items():
|
|
error = UpdateError(field=key, errors=[])
|
|
for field_error in serializer.errors[key]:
|
|
error.errors.append(FieldError(code=field_error.code))
|
|
|
|
errors.append(error)
|
|
|
|
return cls(success=False, errors=errors)
|
|
|
|
|
|
class UpdateAvatar(relay.ClientIDMutation):
|
|
class Input:
|
|
avatar_url = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(UpdateError) # todo: change for consistency
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
avatar_data = kwargs.get("avatar_url")
|
|
|
|
serializer = AvatarUrlSerializer(data={"avatar_url": avatar_data})
|
|
if serializer.is_valid():
|
|
user.avatar_url = avatar_data
|
|
user.save()
|
|
return cls(success=True)
|
|
|
|
errors = []
|
|
for key, value in serializer.errors.items():
|
|
error = UpdateError(field=key, errors=[])
|
|
|
|
for field_error in serializer.errors[key]:
|
|
error.errors.append(FieldError(code=field_error.code))
|
|
|
|
errors.append(error)
|
|
|
|
return cls(success=False, errors=errors)
|
|
|
|
|
|
class UpdateSetting(relay.ClientIDMutation):
|
|
class Input:
|
|
id = graphene.ID(required=True)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
class_id = kwargs.get("id")
|
|
school_class = get_object(SchoolClass, class_id)
|
|
user = info.context.user
|
|
|
|
if school_class and school_class not in user.school_classes.all():
|
|
raise PermissionDenied("Permission denied: Incorrect school class")
|
|
|
|
user.set_selected_class(school_class)
|
|
return cls(success=True)
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(UpdateError)
|
|
|
|
|
|
class JoinClass(relay.ClientIDMutation):
|
|
class Input:
|
|
code = graphene.String(required=True)
|
|
|
|
success = graphene.Boolean()
|
|
school_class = graphene.Field(SchoolClassNode)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
code = kwargs.get("code")
|
|
try:
|
|
school_class = SchoolClass.objects.get(Q(code__iexact=code))
|
|
|
|
if user not in list(school_class.users.all()):
|
|
SchoolClassMember.objects.get_or_create(user=user, school_class=school_class)
|
|
user.set_selected_class(school_class)
|
|
else:
|
|
raise CodeNotFoundException(
|
|
f"[CAJ] Schüler ist bereits in Klasse"
|
|
) # CAJ = Class Already Joined
|
|
|
|
return cls(success=True, school_class=school_class)
|
|
except SchoolClass.DoesNotExist:
|
|
raise CodeNotFoundException(
|
|
f"[CNV] Code ist nicht gültig '{code}'"
|
|
) # CNV = Code Not Valid
|
|
|
|
|
|
class AddRemoveMember(relay.ClientIDMutation):
|
|
class Input:
|
|
member = graphene.ID(required=True)
|
|
school_class = graphene.ID(required=True)
|
|
active = graphene.Boolean(required=True)
|
|
|
|
success = graphene.Boolean()
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
member_id = kwargs.get("member")
|
|
school_class_id = kwargs.get("school_class")
|
|
active = kwargs.get("active")
|
|
user = info.context.user
|
|
|
|
member_pk = from_global_id(member_id)[1]
|
|
school_class = get_object(SchoolClass, school_class_id)
|
|
|
|
if not user.is_teacher() or not school_class.users.filter(pk=user.pk).exists():
|
|
raise PermissionError("Permission denied")
|
|
|
|
school_class_member = SchoolClassMember.objects.get(
|
|
user__pk=member_pk, school_class=school_class
|
|
)
|
|
school_class_member.active = active
|
|
school_class_member.save()
|
|
|
|
return cls(success=True)
|
|
|
|
|
|
class UpdateSchoolClass(TeacherOnlyMutation):
|
|
class Input:
|
|
id = graphene.ID(required=True)
|
|
name = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
school_class = graphene.Field(SchoolClassNode)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
id = kwargs.get("id")
|
|
name = kwargs.get("name")
|
|
user = info.context.user
|
|
|
|
# todo: only allow to edit your own school class
|
|
school_class = get_object(SchoolClass, id)
|
|
school_class.name = name
|
|
school_class.save()
|
|
|
|
return cls(success=True, school_class=school_class)
|
|
|
|
|
|
class CreateSchoolClass(TeacherOnlyMutation):
|
|
class Input:
|
|
name = graphene.String()
|
|
|
|
result = CreateSchoolClassResult()
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
name = kwargs.get("name")
|
|
user = info.context.user
|
|
|
|
try:
|
|
school_class = SchoolClass.objects.create(name=name)
|
|
SchoolClassMember.objects.get_or_create(school_class=school_class, user=user)
|
|
user.set_selected_class(school_class)
|
|
return cls(result=school_class)
|
|
except IntegrityError:
|
|
return cls(result=DuplicateFailure)
|
|
|
|
|
|
class CreateTeam(TeacherOnlyMutation):
|
|
class Input:
|
|
name = graphene.String(required=True)
|
|
|
|
result = CreateTeamResult()
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
name = kwargs.get("name")
|
|
user = info.context.user
|
|
|
|
try:
|
|
team = Team.objects.create(name=name, creator=user)
|
|
team.generate_code()
|
|
user.team = team
|
|
user.save()
|
|
return cls(result=team)
|
|
except IntegrityError:
|
|
return cls(result=DuplicateFailure)
|
|
|
|
|
|
class UpdateTeam(TeacherOnlyMutation):
|
|
class Input:
|
|
id = graphene.ID(required=True)
|
|
name = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
team = graphene.Field(TeamNode)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
id = kwargs.get("id")
|
|
name = kwargs.get("name")
|
|
user = info.context.user
|
|
|
|
team = get_object(Team, id)
|
|
if user not in team.members.all():
|
|
logger.info("User not part of this team")
|
|
raise PermissionError("Permission denied")
|
|
team.name = name
|
|
team.save()
|
|
|
|
return cls(success=True, team=team)
|
|
|
|
|
|
class JoinTeam(TeacherOnlyMutation):
|
|
class Input:
|
|
code = graphene.String(required=True)
|
|
|
|
success = graphene.Boolean()
|
|
team = graphene.Field(TeamNode)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
code = kwargs.get("code")
|
|
try:
|
|
team = Team.objects.get(Q(code__iexact=code))
|
|
user.team = team
|
|
user.save()
|
|
|
|
return cls(success=True, team=team)
|
|
except Team.DoesNotExist:
|
|
raise CodeNotFoundException(
|
|
f"[CNV] Code ist nicht gültig '{code}'"
|
|
) # CNV = Code Not Valid
|
|
|
|
|
|
class LeaveTeam(graphene.Mutation):
|
|
success = graphene.Boolean()
|
|
|
|
@classmethod
|
|
def mutate(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
user.team = None
|
|
user.save()
|
|
return cls(success=True)
|
|
|
|
|
|
class UpdateOnboardingProgress(graphene.Mutation):
|
|
success = graphene.Boolean()
|
|
|
|
@classmethod
|
|
def mutate(cls, root, info, **kwargs):
|
|
user = info.context.user
|
|
if not user.onboarding_visited:
|
|
user.onboarding_visited = True
|
|
user.save()
|
|
|
|
return cls(success=True)
|
|
|
|
|
|
class ProfileMutations:
|
|
update_password = UpdatePassword.Field()
|
|
update_avatar = UpdateAvatar.Field()
|
|
update_setting = UpdateSetting.Field()
|
|
join_class = JoinClass.Field()
|
|
add_remove_member = AddRemoveMember.Field()
|
|
update_school_class = UpdateSchoolClass.Field()
|
|
create_school_class = CreateSchoolClass.Field()
|
|
update_onboarding_progress = UpdateOnboardingProgress.Field()
|
|
create_team = CreateTeam.Field()
|
|
join_team = JoinTeam.Field()
|
|
update_team = UpdateTeam.Field()
|
|
leave_team = LeaveTeam.Field()
|