269 lines
13 KiB
Python
269 lines
13 KiB
Python
|
from django.contrib.auth import authenticate, login, logout
|
||
|
|
from django.db.models import Q
|
||
|
|
|
||
|
|
from rest_framework.status import HTTP_204_NO_CONTENT, HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND, HTTP_200_OK, HTTP_201_CREATED
|
||
|
|
from rest_framework.decorators import action
|
||
|
|
from rest_framework.permissions import AllowAny, IsAuthenticated, IsAuthenticatedOrReadOnly
|
||
|
|
from rest_framework.response import Response
|
||
|
|
from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
|
||
|
|
|
||
|
|
from apps.accounts.models import Invite, Organization, Role, User
|
||
|
|
from apps.accounts.serializers import InviteSerializer, OrganizationSerializer, RoleSerializer, UserSerializer
|
||
|
|
|
||
|
|
class UserViewSet(ReadOnlyModelViewSet):
|
||
|
|
|
||
|
|
queryset = User.objects.all()
|
||
|
|
serializer_class = UserSerializer
|
||
|
|
permission_classes = [IsAuthenticatedOrReadOnly]
|
||
|
|
lookup_field = 'uuid'
|
||
|
|
|
||
|
|
@action(detail=False, methods=['post'], permission_classes=[AllowAny])
|
||
|
|
def login(self, request):
|
||
|
|
email_address = request.data.get('email_address')
|
||
|
|
password = request.data.get('password')
|
||
|
|
if not email_address or not password:
|
||
|
|
return Response({'error': 'Email and password are required'}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
email_address = User.objects.normalize_email(email_address)
|
||
|
|
user = authenticate(request, username=email_address, password=password)
|
||
|
|
if user is None:
|
||
|
|
return Response({'error': 'Invalid credentials'}, status=HTTP_401_UNAUTHORIZED)
|
||
|
|
|
||
|
|
login(request, user)
|
||
|
|
return Response({'user': UserSerializer(user).data, 'message': 'Login successful', 'success': True}, status=HTTP_200_OK)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['post'], permission_classes=[IsAuthenticated])
|
||
|
|
def logout(self, request):
|
||
|
|
logout(request)
|
||
|
|
return Response({'message': 'Logout successful', 'success': True}, status=HTTP_200_OK)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['get'], permission_classes=[IsAuthenticated])
|
||
|
|
def me(self, request):
|
||
|
|
user_data = UserSerializer(request.user).data
|
||
|
|
user_data['success'] = True
|
||
|
|
return Response(user_data)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['get'], permission_classes=[AllowAny])
|
||
|
|
def session(self, request):
|
||
|
|
return Response({'isAuthenticated': request.user.is_authenticated, 'isStaff': request.user.is_staff if request.user.is_authenticated else False})
|
||
|
|
|
||
|
|
@action(detail=False, methods=['post'], permission_classes=[AllowAny])
|
||
|
|
def signup(self, request):
|
||
|
|
try:
|
||
|
|
data = request.data
|
||
|
|
except:
|
||
|
|
return Response({'detail': 'Invalid data provided.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
email_address = data.get('email_address')
|
||
|
|
if not email_address:
|
||
|
|
return Response({'detail': 'Email address is required.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
email_address = User.objects.normalize_email(email_address)
|
||
|
|
if User.objects.filter(email_address=email_address).exists():
|
||
|
|
return Response({'detail': 'Email address already exists.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
if not data.get('first_name') or not data.get('last_name'):
|
||
|
|
return Response({'detail': 'First and last name(s) must be provided.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
if type(manager:=data.get('manager')) is not bool:
|
||
|
|
if manager in ['true', 'True']:
|
||
|
|
manager = True
|
||
|
|
elif manager in ['false', 'False']:
|
||
|
|
manager = False
|
||
|
|
else:
|
||
|
|
return Response({'detail': '"manager" field must be a boolean value.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
if data.get('password') != data.get('confirm_password'):
|
||
|
|
return Response({'detail': 'Passwords do not match.', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
try:
|
||
|
|
user = User.objects.create_user(
|
||
|
|
email_address=email_address,
|
||
|
|
password=data.get('password'),
|
||
|
|
first_name=data.get('first_name'),
|
||
|
|
last_name=data.get('last_name'),
|
||
|
|
date_of_birth=data.get('date_of_birth'),
|
||
|
|
is_manager=manager,
|
||
|
|
)
|
||
|
|
return Response({'detail': 'User account created successfully.', 'success': True}, status=HTTP_201_CREATED)
|
||
|
|
except Exception as e:
|
||
|
|
return Response({'detail': str(e), 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['post'], permission_classes=[IsAuthenticated])
|
||
|
|
def change_password(self, request):
|
||
|
|
data = request.data
|
||
|
|
required_fields = ['old_password', 'password', 'confirm_password']
|
||
|
|
for field in required_fields:
|
||
|
|
if not data.get(field):
|
||
|
|
return Response({'detail': f'"{field}" not provided', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
if data.get('password') != data.get('confirm_password'):
|
||
|
|
return Response({'detail': 'Passwords do not match', 'success': False}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
user = request.user
|
||
|
|
if not user.check_password(data.get('old_password')):
|
||
|
|
return Response({'detail': 'Old password is incorrect', 'success': False}, status=HTTP_401_UNAUTHORIZED)
|
||
|
|
user.set_password(data.get('password'))
|
||
|
|
user.save()
|
||
|
|
return Response({'detail': 'Password changed successfully', 'success': True}, status=HTTP_200_OK)
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class OrganizationViewSet(ModelViewSet):
|
||
|
|
queryset = Organization.objects.all()
|
||
|
|
serializer_class = OrganizationSerializer
|
||
|
|
permission_classes = [IsAuthenticated]
|
||
|
|
lookup_field = 'uuid'
|
||
|
|
|
||
|
|
def get_queryset(self):
|
||
|
|
return Organization.objects.filter(
|
||
|
|
Q(owner=self.request.user) | Q(members=self.request.user)
|
||
|
|
).distinct()
|
||
|
|
|
||
|
|
def perform_create(self, serializer):
|
||
|
|
organization = serializer.save(owner=self.request.user)
|
||
|
|
organization.members.add(self.request.user)
|
||
|
|
|
||
|
|
def update(self, request, *args, **kwargs):
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Forbidden'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
return super().update(request, *args, **kwargs)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['get'], url_path='invite')
|
||
|
|
def list_invites(self, request, uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
invites = organization.invites.all()
|
||
|
|
serializer = InviteSerializer(invites, many=True, context={'request': request})
|
||
|
|
return Response(serializer.data)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['post'], url_path='create-invite')
|
||
|
|
def create_invite(self, request, uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Forbidden'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
max_uses = request.query_params.get('max_uses') or request.data.get('max_uses', 1)
|
||
|
|
invitation = Invite.objects.create(
|
||
|
|
organization=organization,
|
||
|
|
created_by=request.user,
|
||
|
|
max_uses=int(max_uses) if str(max_uses).isdigit() else 1
|
||
|
|
)
|
||
|
|
return Response(InviteSerializer(invitation, context={'request': request}).data)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['delete'], url_path=r'revoke-invite/(?P<token>[0-9a-f-]{36})')
|
||
|
|
def revoke_invite(self, request, uuid=None, token=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Only managers can revoke invites'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
invite = organization.invites.filter(token=token).first()
|
||
|
|
if not invite:
|
||
|
|
return Response({'error': 'Invalid invitation token or not found in this organization'}, status=HTTP_404_NOT_FOUND)
|
||
|
|
invite.is_active = False
|
||
|
|
invite.save()
|
||
|
|
return Response({'message': 'Invitation successfully revoked'}, status=HTTP_200_OK)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['post'], url_path='join/(?P<token>[0-9a-f-]{36})')
|
||
|
|
def join(self, request, token=None):
|
||
|
|
try:
|
||
|
|
invitation = Invite.objects.get(token=token)
|
||
|
|
except Invite.DoesNotExist:
|
||
|
|
return Response({'error': 'Not Found'}, status=HTTP_404_NOT_FOUND)
|
||
|
|
|
||
|
|
if not invitation.is_valid():
|
||
|
|
return Response({'error': 'Invalid or expired token'}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
|
||
|
|
organization = invitation.organization
|
||
|
|
if organization.members.filter(id=request.user.id).exists():
|
||
|
|
return Response({'error': 'Already a member'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
organization.members.add(request.user)
|
||
|
|
|
||
|
|
invitation.uses += 1
|
||
|
|
if invitation.uses >= invitation.max_uses:
|
||
|
|
invitation.is_active = False
|
||
|
|
invitation.save()
|
||
|
|
|
||
|
|
return Response({
|
||
|
|
'message': 'Joined',
|
||
|
|
'organization': OrganizationSerializer(organization).data
|
||
|
|
})
|
||
|
|
|
||
|
|
@action(detail=True, methods=['post'], url_path='leave')
|
||
|
|
def leave(self, request, uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
if organization.owner == request.user:
|
||
|
|
return Response({'error': 'Owner cannot leave'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
if not organization.members.filter(id=request.user.id).exists():
|
||
|
|
return Response({'error': 'Not a member'}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
|
||
|
|
organization.members.remove(request.user)
|
||
|
|
return Response({'message': 'Left organization'})
|
||
|
|
|
||
|
|
@action(detail=True, methods=['get'], url_path='members')
|
||
|
|
def list_members(self, request, uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
serializer = UserSerializer(organization.members.all(), many=True)
|
||
|
|
return Response(serializer.data)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['post'], url_path=r'member/(?P<user_id>\d+)/remove')
|
||
|
|
def remove_member(self, request, uuid=None, user_id=None):
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Forbidden'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
organization = self.get_object()
|
||
|
|
if str(organization.owner.id) == str(user_id):
|
||
|
|
return Response({'error': 'Cannot remove owner'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
user_to_remove = organization.members.filter(id=user_id).first()
|
||
|
|
if not user_to_remove:
|
||
|
|
return Response({'error': 'Not found'}, status=HTTP_404_NOT_FOUND)
|
||
|
|
|
||
|
|
organization.members.remove(user_to_remove)
|
||
|
|
return Response({'message': 'Removed'})
|
||
|
|
|
||
|
|
@action(detail=True, methods=['get', 'post'], url_path='role')
|
||
|
|
def roles(self, request, uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
if request.method == 'GET':
|
||
|
|
return Response(RoleSerializer(organization.roles.all(), many=True).data)
|
||
|
|
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Forbidden'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
name = (request.data.get('name') or '').strip()
|
||
|
|
description = (request.data.get('description') or '').strip()
|
||
|
|
|
||
|
|
if not name:
|
||
|
|
return Response({'error': 'Role name is required'}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
|
||
|
|
if organization.roles.filter(name__iexact=name).exists():
|
||
|
|
return Response({'error': 'A role with this name already exists in this organization'}, status=HTTP_400_BAD_REQUEST)
|
||
|
|
|
||
|
|
role = Role.objects.create(name=name, description=description, organization=organization)
|
||
|
|
return Response(RoleSerializer(role).data, status=HTTP_201_CREATED)
|
||
|
|
|
||
|
|
@action(detail=False, methods=['get'], url_path='role/mine')
|
||
|
|
def my_roles(self, request):
|
||
|
|
roles = Role.objects.filter(members=request.user).distinct()
|
||
|
|
serializer = RoleSerializer(roles, many=True)
|
||
|
|
return Response(serializer.data)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['delete'], url_path='role/(?P<role_uuid>[0-9a-f-]{36})')
|
||
|
|
def delete_role(self, request, uuid=None, role_uuid=None):
|
||
|
|
if not request.user.is_manager:
|
||
|
|
return Response({'error': 'Forbidden'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
role = Role.objects.filter(uuid=role_uuid, organization__uuid=uuid)
|
||
|
|
if not role.exists():
|
||
|
|
return Response({'error': 'Not found'}, status=HTTP_404_NOT_FOUND)
|
||
|
|
|
||
|
|
role.delete()
|
||
|
|
return Response(status=HTTP_204_NO_CONTENT)
|
||
|
|
|
||
|
|
@action(detail=True, methods=['post'], url_path='role/(?P<role_uuid>[0-9a-f-]{36})/join')
|
||
|
|
def join_role(self, request, uuid=None, role_uuid=None):
|
||
|
|
organization = self.get_object()
|
||
|
|
|
||
|
|
role = Role.objects.filter(uuid=role_uuid, organization=organization).first()
|
||
|
|
if not role:
|
||
|
|
return Response({'error': 'Role not found'}, status=HTTP_404_NOT_FOUND)
|
||
|
|
|
||
|
|
if not organization.members.filter(id=request.user.id).exists() and organization.owner != request.user:
|
||
|
|
return Response({'error': 'Not a member of this organization'}, status=HTTP_403_FORBIDDEN)
|
||
|
|
|
||
|
|
if role.members.filter(id=request.user.id).exists():
|
||
|
|
return Response({'message': 'Already a member of this role'}, status=HTTP_200_OK)
|
||
|
|
|
||
|
|
role.members.add(request.user)
|
||
|
|
return Response({'message': 'Joined role successfully'}, status=HTTP_200_OK)
|