person.py

from django.db.models import Q
from rest_framework import mixins
from rest_framework import status
from rest_framework.authentication import SessionAuthentication, BasicAuthentication
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet

from greg.models import Identity, Person
from greg.permissions import IsSponsor
from greg.utils import is_identity_duplicate
from gregui import validation
from gregui.api.serializers.IdentityDuplicateError import IdentityDuplicateError
from gregui.api.serializers.guest import GuestSerializer, create_identity_or_update
from gregui.models import GregUserProfile


class PersonViewSet(mixins.RetrieveModelMixin, mixins.UpdateModelMixin, GenericViewSet):
    """
    Fetch person info for any guest as long as you are a sponsor

    This is required for the functionality where a sponsor wants to add a guest role to
    an already existing person that is not already their guest.

    Returns enough information to fill a profile page in the frontend
    """

    authentication_classes = [SessionAuthentication, BasicAuthentication]
    permission_classes = [IsAuthenticated, IsSponsor]
    queryset = Person.objects.all()
    http_methods = ["get", "patch"]
    serializer_class = GuestSerializer

    def update(self, request, *args, **kwargs):
        """
        Overriding the update-method to be able to return an error response
        with a special format if there is a duplicate identity
        """
        partial = kwargs.pop("partial", False)
        instance = self.get_object()
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)

        try:
            self.perform_update(serializer)
        except IdentityDuplicateError as error:
            return Response(
                status=status.HTTP_400_BAD_REQUEST,
                data={
                    "code": error.code,
                    "message": error.detail["message"],
                },
            )

        if getattr(instance, "_prefetched_objects_cache", None):
            # If 'prefetch_related' has been applied to a queryset, we need to
            # forcibly invalidate the prefetch cache on the instance.
            instance._prefetched_objects_cache = {}

        return Response(serializer.data)

    def perform_update(self, serializer):
        """Update email when doing patch"""
        email = self.request.data.get("email")
        person = self.get_object()
        validation.validate_email(email)

        if is_identity_duplicate(Identity.IdentityType.PRIVATE_EMAIL, email):
            raise IdentityDuplicateError(
                Identity.IdentityType.PRIVATE_EMAIL, "duplicate_private_email"
            )

        create_identity_or_update(Identity.IdentityType.PRIVATE_EMAIL, email, person)
        return super().perform_update(serializer)


class PersonSearchViewSet(GenericViewSet):
    """Search for persons using name, email, phone number and birth date"""

    authentication_classes = [SessionAuthentication, BasicAuthentication]
    permission_classes = [IsAuthenticated, IsSponsor]

    def list(self, request, *args, **kwargs):
        if "q" not in self.request.query_params:
            return Response(
                status=status.HTTP_400_BAD_REQUEST,
                data={
                    "code": "no_search_term",
                    "message": "No search query parameter given",
                },
            )

        if len(self.request.query_params["q"]) > 50:
            return Response(
                status=status.HTTP_400_BAD_REQUEST,
                data={
                    "code": "search_term_too_large",
                    "message": "Search term is too large",
                },
            )

        hits = self.get_hits()
        return Response(hits)

    def get_hits(self):
        search = self.request.query_params["q"]

        split_search = search.split()
        words_joined = "|".join(map(str, split_search))
        # Create a regex with the terms in the search or-ed together. This will trigger a match
        # if one of the fields that are being searched contains one of the terms
        search_regex = r"^(%s)" % words_joined

        hits = []
        # First look for hits on name and birth date
        persons = Person.objects.filter(
            Q(first_name__iregex=search_regex)
            | Q(last_name__iregex=search_regex)
            | Q(date_of_birth__iregex=search_regex)
        )[:10]

        included_persons = []
        for person in persons:
            hits.append(
                {"pid": person.id, "first": person.first_name, "last": person.last_name}
            )
            included_persons.append(person.id)

        if len(hits) == 10:
            # Max number of hits, no need to search more
            return hits

        # Look for hits in e-mail and mobile phone
        identities = Identity.objects.filter(
            value__iregex=search_regex,
            type__in=[
                Identity.IdentityType.PRIVATE_EMAIL,
                Identity.IdentityType.PRIVATE_MOBILE_NUMBER,
            ],
        )[: (10 - len(hits))]

        for identity in identities:
            if identity.person_id in included_persons:
                continue

            hits.append(
                {
                    "pid": identity.person_id,
                    "first": identity.person.first_name,
                    "last": identity.person.last_name,
                    "value": identity.value,
                    "type": identity.type,
                }
            )

        return hits


class GuestInfoViewSet(mixins.ListModelMixin, GenericViewSet):
    """
    Fetch all guests at the sponsor's units.

    If the Sponsor's connection to the Unit is marked with hierarchical access, guests
    at child units of the Unit are included.

    Lists all persons connected to the roles the logged in sponsor is connected to.
    """

    authentication_classes = [SessionAuthentication, BasicAuthentication]
    permission_classes = [IsAuthenticated, IsSponsor]
    serializer_class = GuestSerializer

    def get_queryset(self):
        """
        Fetch Persons connected to the sponsor some way.

        Any person with a role connected to the same unit as the sponsor, or a unit
        that the sponsor is connected to through hierarchical access.
        """

        user = GregUserProfile.objects.get(user=self.request.user)
        units = user.sponsor.get_allowed_units()
        return (
            Person.objects.filter(roles__orgunit__in=list(units))
            .distinct()
            .order_by("id")
        )