# ~*~ coding: utf-8 ~*~ from django.core.cache import cache from django.utils.translation import ugettext as _ from rest_framework.decorators import action from rest_framework import generics from rest_framework.response import Response from rest_framework_bulk import BulkModelViewSet from common.db.aggregates import GroupConcat from common.permissions import ( IsOrgAdmin, IsOrgAdminOrAppUser, CanUpdateDeleteUser, IsSuperUser ) from common.mixins import CommonApiMixin from common.utils import get_logger from orgs.utils import current_org from orgs.models import ROLE as ORG_ROLE, OrganizationMember from .. import serializers from ..serializers import UserSerializer, UserRetrieveSerializer, MiniUserSerializer, InviteSerializer from .mixins import UserQuerysetMixin from ..models import User from ..signals import post_user_create from ..filters import OrgRoleUserFilterBackend logger = get_logger(__name__) __all__ = [ 'UserViewSet', 'UserChangePasswordApi', 'UserUnblockPKApi', 'UserResetOTPApi', ] class UserViewSet(CommonApiMixin, UserQuerysetMixin, BulkModelViewSet): filter_fields = ('username', 'email', 'name', 'id', 'source') search_fields = filter_fields permission_classes = (IsOrgAdmin, CanUpdateDeleteUser) serializer_classes = { 'default': UserSerializer, 'retrieve': UserRetrieveSerializer, 'suggestion': MiniUserSerializer, 'invite': InviteSerializer, } extra_filter_backends = [OrgRoleUserFilterBackend] def get_queryset(self): return super().get_queryset().annotate( gc_m2m_org_members__role=GroupConcat('m2m_org_members__role'), ).prefetch_related('groups') def send_created_signal(self, users): if not isinstance(users, list): users = [users] for user in users: post_user_create.send(self.__class__, user=user) @staticmethod def set_users_to_org(users, org_roles, update=False): # 只有真实存在的组织才真正关联用户 if not current_org or not current_org.is_real(): return for user, roles in zip(users, org_roles): if update and roles is None: continue if not roles: # 当前组织创建的用户，至少是该组织的`User` roles = [ORG_ROLE.USER] OrganizationMember.objects.set_user_roles(current_org, user, roles) def perform_create(self, serializer): org_roles = self.get_serializer_org_roles(serializer) # 创建用户 users = serializer.save() if isinstance(users, User): users = [users] self.set_users_to_org(users, org_roles) self.send_created_signal(users) def get_permissions(self): if self.action in ["retrieve", "list"]: self.permission_classes = (IsOrgAdminOrAppUser,) if self.request.query_params.get('all'): self.permission_classes = (IsSuperUser,) return super().get_permissions() def perform_destroy(self, instance): if current_org.is_real(): instance.remove() else: return super().perform_destroy(instance) def perform_bulk_destroy(self, objects): for obj in objects: self.check_object_permissions(self.request, obj) self.perform_destroy(obj) @staticmethod def get_serializer_org_roles(serializer): validated_data = serializer.validated_data # `org_roles` 先 `pop` if isinstance(validated_data, list): org_roles = [item.pop('org_roles', None) for item in validated_data] else: org_roles = [validated_data.pop('org_roles', None)] return org_roles def perform_update(self, serializer): org_roles = self.get_serializer_org_roles(serializer) users = serializer.save() if isinstance(users, User): users = [users] self.set_users_to_org(users, org_roles, update=True) def perform_bulk_update(self, serializer): # TODO: 需要测试 users_ids = [ d.get("id") or d.get("pk") for d in serializer.validated_data ] users = current_org.get_members().filter(id__in=users_ids) for user in users: self.check_object_permissions(self.request, user) return super().perform_bulk_update(serializer) @action(methods=['get'], detail=False, permission_classes=(IsOrgAdmin,)) def suggestion(self, request): queryset = User.objects.exclude(role=User.ROLE.APP) queryset = self.filter_queryset(queryset) queryset = queryset[:3] page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response(serializer.data) @action(methods=['post'], detail=False, permission_classes=(IsOrgAdmin,)) def invite(self, request): data = request.data if not isinstance(data, list): data = [request.data] if not current_org or not current_org.is_real(): error = {"error": "Not a valid org"} return Response(error, status=400) serializer_cls = self.get_serializer_class() serializer = serializer_cls(data=data, many=True) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data for i in validated_data: i['org_id'] = current_org.org_id() relations = [OrganizationMember(**i) for i in validated_data] OrganizationMember.objects.bulk_create(relations, ignore_conflicts=True) return Response(serializer.data, status=201) class UserChangePasswordApi(UserQuerysetMixin, generics.RetrieveUpdateAPIView): permission_classes = (IsOrgAdmin,) serializer_class = serializers.ChangeUserPasswordSerializer def perform_update(self, serializer): user = self.get_object() user.password_raw = serializer.validated_data["password"] user.save() class UserUnblockPKApi(UserQuerysetMixin, generics.UpdateAPIView): permission_classes = (IsOrgAdmin,) serializer_class = serializers.UserSerializer key_prefix_limit = "_LOGIN_LIMIT_{}_{}" key_prefix_block = "_LOGIN_BLOCK_{}" def perform_update(self, serializer): user = self.get_object() username = user.username if user else '' key_limit = self.key_prefix_limit.format(username, '*') key_block = self.key_prefix_block.format(username) cache.delete_pattern(key_limit) cache.delete(key_block) class UserResetOTPApi(UserQuerysetMixin, generics.RetrieveAPIView): permission_classes = (IsOrgAdmin,) serializer_class = serializers.ResetOTPSerializer def retrieve(self, request, *args, **kwargs): user = self.get_object() if kwargs.get('pk') else request.user if user == request.user: msg = _("Could not reset self otp, use profile reset instead") return Response({"error": msg}, status=401) if user.mfa_enabled: user.reset_mfa() user.save() return Response({"msg": "success"})