159 lines
6.6 KiB
Python
159 lines
6.6 KiB
Python
from django.db import transaction
|
||
from django.shortcuts import get_object_or_404
|
||
from rest_framework import serializers
|
||
from rest_framework.exceptions import APIException, PermissionDenied
|
||
from rest_framework.permissions import IsAuthenticated
|
||
from rest_framework.response import Response
|
||
from rest_framework.views import APIView
|
||
|
||
from .models import DelegatedPermission, PermissionDefaults, UserPermissions, User
|
||
from .permissions import (
|
||
MANAGEMENT_PERMISSION_NAMES,
|
||
PERMISSION_NAMES,
|
||
default_permissions,
|
||
effective_permissions,
|
||
is_admin,
|
||
management_permissions,
|
||
)
|
||
|
||
|
||
class PermissionValues(serializers.Serializer):
|
||
request_leave = serializers.BooleanField(allow_null=True)
|
||
create_personal_event = serializers.BooleanField(allow_null=True)
|
||
create_publication = serializers.BooleanField(allow_null=True)
|
||
|
||
|
||
class ResponsibilityValue(serializers.Serializer):
|
||
enabled = serializers.BooleanField(default=False)
|
||
all_users = serializers.BooleanField(default=False)
|
||
user_ids = serializers.ListField(
|
||
child=serializers.IntegerField(min_value=1),
|
||
allow_empty=True,
|
||
default=list,
|
||
)
|
||
|
||
def validate(self, attrs):
|
||
if attrs['enabled'] and not attrs['all_users'] and not attrs['user_ids']:
|
||
raise serializers.ValidationError('Choisissez au moins un salarié ou activez « Toute l’équipe ».')
|
||
return attrs
|
||
|
||
|
||
class ResponsibilityValues(serializers.Serializer):
|
||
approve_requests = ResponsibilityValue(required=False)
|
||
manage_schedules = ResponsibilityValue(required=False)
|
||
view_statistics = ResponsibilityValue(required=False)
|
||
|
||
|
||
class PermissionInput(serializers.Serializer):
|
||
scope = serializers.ChoiceField(choices=['team', 'user'])
|
||
user_id = serializers.IntegerField(min_value=1, required=False)
|
||
expected_revision = serializers.IntegerField(min_value=0)
|
||
permissions = PermissionValues()
|
||
responsibilities = ResponsibilityValues(required=False)
|
||
reset_individual = serializers.BooleanField(default=False)
|
||
|
||
def validate(self, attrs):
|
||
if attrs['scope'] == 'team' and any(v is None for v in attrs['permissions'].values()):
|
||
raise serializers.ValidationError('Choisissez Autorisé ou Non autorisé pour chaque règle de l’équipe.')
|
||
if attrs['scope'] == 'user' and (not attrs.get('user_id') or attrs['reset_individual']):
|
||
raise serializers.ValidationError('Choisissez le salarié concerné.')
|
||
if attrs['scope'] == 'team' and attrs.get('responsibilities'):
|
||
raise serializers.ValidationError('Les responsabilités de gestion sont attribuées individuellement.')
|
||
return attrs
|
||
|
||
|
||
class PermissionConflict(APIException):
|
||
status_code = 409
|
||
default_detail = 'Les autorisations ont changé. Actualisez cette page avant d’enregistrer.'
|
||
|
||
|
||
def configuration():
|
||
defaults = default_permissions()
|
||
state = PermissionDefaults.objects.filter(pk=1).first()
|
||
users = []
|
||
for user in User.objects.select_related('feature_permissions').order_by('first_name', 'last_name', 'username'):
|
||
try:
|
||
override = user.feature_permissions
|
||
except UserPermissions.DoesNotExist:
|
||
override = None
|
||
users.append(dict(
|
||
id=user.pk,
|
||
username=user.username,
|
||
first_name=user.first_name,
|
||
last_name=user.last_name,
|
||
is_admin=is_admin(user),
|
||
is_active=user.is_active,
|
||
overrides={key: getattr(override, key, None) for key in PERMISSION_NAMES},
|
||
permissions=effective_permissions(user, defaults),
|
||
responsibilities=management_permissions(user),
|
||
))
|
||
return dict(defaults=defaults, revision=state.revision if state else 0, users=users)
|
||
|
||
|
||
class PermissionConfigurationView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def initial(self, request, *args, **kwargs):
|
||
super().initial(request, *args, **kwargs)
|
||
if not is_admin(request.user):
|
||
raise PermissionDenied('Seuls les administrateurs peuvent gérer les autorisations.')
|
||
|
||
@transaction.atomic
|
||
def get(self, request):
|
||
PermissionDefaults.objects.get_or_create(pk=1)
|
||
PermissionDefaults.objects.select_for_update().get(pk=1)
|
||
return Response(configuration())
|
||
|
||
@transaction.atomic
|
||
def put(self, request):
|
||
serializer = PermissionInput(data=request.data)
|
||
serializer.is_valid(raise_exception=True)
|
||
values = serializer.validated_data
|
||
PermissionDefaults.objects.get_or_create(pk=1)
|
||
state = PermissionDefaults.objects.select_for_update().get(pk=1)
|
||
if state.revision != values['expected_revision']:
|
||
raise PermissionConflict()
|
||
|
||
if values['scope'] == 'team':
|
||
for key, value in values['permissions'].items():
|
||
setattr(state, key, value)
|
||
if values['reset_individual']:
|
||
UserPermissions.objects.all().update(**dict.fromkeys(PERMISSION_NAMES, None))
|
||
else:
|
||
user = get_object_or_404(User, pk=values['user_id'])
|
||
if is_admin(user):
|
||
raise serializers.ValidationError('Les administrateurs ont déjà tous les accès de gestion.')
|
||
|
||
UserPermissions.objects.update_or_create(user=user, defaults=values['permissions'])
|
||
|
||
responsibilities = values.get('responsibilities', {})
|
||
for code in MANAGEMENT_PERMISSION_NAMES:
|
||
if code not in responsibilities:
|
||
continue
|
||
rule = responsibilities[code]
|
||
if not rule['enabled']:
|
||
DelegatedPermission.objects.filter(user=user, code=code).delete()
|
||
continue
|
||
|
||
target_ids = set(rule.get('user_ids', []))
|
||
if not rule['all_users']:
|
||
existing_ids = set(User.objects.filter(pk__in=target_ids, is_active=True).values_list('pk', flat=True))
|
||
missing = target_ids - existing_ids
|
||
if missing:
|
||
raise serializers.ValidationError({
|
||
'responsibilities': f'Un salarié sélectionné pour « {code} » est introuvable ou désactivé.'
|
||
})
|
||
row, _created = DelegatedPermission.objects.update_or_create(
|
||
user=user,
|
||
code=code,
|
||
defaults={'all_users': rule['all_users']},
|
||
)
|
||
if rule['all_users']:
|
||
row.targets.clear()
|
||
else:
|
||
row.targets.set(sorted(target_ids))
|
||
|
||
state.revision += 1
|
||
state.save()
|
||
return Response(configuration())
|