Представления

Представления - основная логика нашего сайта, именно здесь мы будем создавать наши эндпоинты. Для этого перейдём в файл views.py в наших приложениях.

Для большинства функций написан длинный декоратор @extend_schema, он используется для конкретизации документации Swagger. Об этом подробнее здесь.

oidcsurvey

АвторизацияГруппа

from urllib.parse import urlencode, quote

from allauth.socialaccount.providers.oauth2.client import OAuth2Client
from allauth.socialaccount.providers.openid_connect.views import OpenIDConnectAdapter
from dj_rest_auth.registration.views import SocialLoginView
from django.conf import settings
from django.http import HttpResponseRedirect
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import extend_schema, inline_serializer, OpenApiParameter
from rest_framework import status, serializers
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView


BE_BASE_URL = settings.BE_BASE_URL
FE_BASE_URL = settings.FE_BASE_URL


class ItmoIdAuthUrl(APIView):
    @extend_schema(
        summary="Return ITMO.ID authentication URL",
        responses=inline_serializer(
            name="ItmoIdAuthUrl",
            fields={"url": serializers.URLField()},
        ),
    )
    def get(self, request: Request):
        query_params = {
            "client_id": settings.ITMO_ID_CLIENT_ID,
            "response_type": "code",
            "redirect_uri": f"{BE_BASE_URL}/oidc/auth",
            "scope": "openid name edu",
        }

        url = f"https://id.itmo.pro/auth/realms/itmo/protocol/openid-connect/auth?{urlencode(query_params)}"

        return Response({"url": url}, status=status.HTTP_200_OK)


class ItmoIdAuth(APIView):
    @extend_schema(
        summary="Get 'code' from ITMO.ID and return it with redirect to FE app",
        parameters=[OpenApiParameter(name="code", type=OpenApiTypes.STR)],
        responses={
            status.HTTP_400_BAD_REQUEST: inline_serializer(
                name="ItmoIdAuthError",
                fields={"detail": serializers.CharField()},
            ),
        },
    )
    def get(self, request: Request):
        auth_code = request.query_params.get("code")
        if not auth_code:
            return Response({"detail": "Missing 'code' parameter"}, status=status.HTTP_400_BAD_REQUEST)

        redirect_url = f"{FE_BASE_URL}/login?code={quote(auth_code)}"
        return HttpResponseRedirect(redirect_url)


class ItmoIdAdapter(OpenIDConnectAdapter):
    def __init__(self, request: Request):
        provider_id = "itmo_id"
        super().__init__(request, provider_id)


class ItmoIdLogin(SocialLoginView):
    """Get 'code' and request ITMO.ID for access token and userinfo. Create new User and ITMO.ID Profile"""

    adapter_class = ItmoIdAdapter
    callback_url = f"{BE_BASE_URL}/oidc/auth"
    client_class = OAuth2Client

В начале файла импортируем из настроек базовые URL для наших бэкенда и фронтенда, они нам понадобятся для редиректов.

В классе ItmoIdAuthUrl создадим метод get, который будет создавать и возвращать ссылку на авторизацию в ITMO.ID на основе данных имеющегося клиента.

Класс ItmoIdAdapter - наследник от OpenIDConnectAdapter (из библиотеки django-allauth), который переопределяет только его метод инициализации, дополнительно указывая название используемого OIDC провайдера.

В классе ItmoIdAuth создадим метод get, который из query параметров получит code и отправит его во фронтенд приложение. Эндпоинт, связанный с этим классом, указывается, как redirect_uri в настройках провайдера OIDC, и после успешной авторизации в нём, он редиректнет пользователя на него вместе с необходимыми параметрами.

Класс ItmoIdLogin - наследник от SocialLoginView (из библиотеки dj-rest-auth). Именно он отвечает за основной процесс авторизации в стороннем провайдере: обмен кода авторизации на JWT токены провайдера, получение из токена информации о пользователе, создание нового пользователя в БД, создание "социального аккаунта" пользователя (со всей полученной информацией), выдача собственных JWT токенов.

Также, для удобной работы создадим Django-сигнал, который при создании социального аккаунта так же создаст запись в нашей таблице профилей ITMO.ID, и привяжет эту запись к нужному пользователю:

signals.py

import logging

from allauth.socialaccount.models import SocialAccount
from django.db import DatabaseError
from django.db.models.signals import post_save
from django.dispatch import receiver

from apps.oidc.models import ItmoIdProfile

logger = logging.getLogger("root")


@receiver(post_save, sender=SocialAccount)
def itmo_id_profile_create(sender, instance: SocialAccount, created: bool, **kwargs):
    if not created or instance.provider != "itmo_id":
        return

    data = instance.extra_data
    group = data.get("groups", [{}])[-1]

    try:
        ItmoIdProfile.objects.create(
            user_id=instance.user_id,
            isu=data.get("isu"),
            course=group.get("course"),
            faculty=group.get("faculty", {}).get("short_name", ""),
            group=group.get("name", ""),
        )
    except DatabaseError as exc:
        logger.exception(exc)

Этот сигнал получает информацию из полученного от ITMO.ID токена (сохранённую в социальном аккаунте в виде словаря) и распрасит её в поля модели, создав запись в ней.

Чтобы он работал, его нужно импортировать при инициализации приложения (в методе ready):

apps.py

from django.apps import AppConfig


class AuthConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "apps.oidc"
    verbose_name = "Авторизация"

    def ready(self):
        from . import signals  # noqa

from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import extend_schema, inline_serializer
from rest_framework import status, serializers
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView


class ItmoIdProfileGroupView(APIView):
    permission_classes = [IsAuthenticated]

    @extend_schema(
        summary="Get user's education group from ITMO.ID",
        responses={
            status.HTTP_400_BAD_REQUEST: inline_serializer(
                name="ItmoIdProfileGroup",
                fields={"group": serializers.CharField()},
            ),
            status.HTTP_404_NOT_FOUND: inline_serializer(
                name="ItmoIdProfileGroupNotFound", fields={"error": serializers.CharField()}
            ),
        },
    )
    def get(self, request: Request):
        user = request.user
        if not hasattr(user, "profile"):
            return Response({"error": "Profile does not exist"}, status.HTTP_404_NOT_FOUND)

        group = user.profile.group
        return Response({"group": group})

В этом же приложении и в этом же файле также создадим класс ItmoIdProfileGroupView. В нём метод get будет возвращать учебную группу пользователя. Для этого класса нужно указать доступ только для авторизованных пользователей с помощью permission_classes = [IsAuthenticated].

ВопросыПроверкаОтветы

from rest_framework import generics
from rest_framework.permissions import IsAuthenticated

from .serializers import SurveyQuestionSerializer
from .models import SurveyQuestion


class SurveyQuestionList(generics.ListAPIView):
    """Get survey's questions with all needed information"""

    queryset = SurveyQuestion.objects.order_by("order").all()
    serializer_class = SurveyQuestionSerializer
    permission_classes = [IsAuthenticated]

    def get_serializer_context(self):
        return {"request": self.request}

В этом классе за основу возьму generics.ListAPIView, возвращающий список объектов. Укажем необходимый запрос с сортировкой по полю order, ранее созданный сериализатор и доступ только авторизованным пользователем. Также в методе get_serializer_context передадим полученный запрос для корректной работы нашего сериализатора.

from drf_spectacular.utils import extend_schema, inline_serializer
from rest_framework import status, serializers
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import SurveyFaculty, SurveyGroup, SurveyAnswer


class SurveyCheckUserView(APIView):
    permission_classes = [IsAuthenticated]

    @extend_schema(
        summary="Get the user's suitability to take the survey",
        responses={
            status.HTTP_200_OK: inline_serializer(
                name="SurveyCheckUser",
                fields={
                    "access": serializers.BooleanField(),
                    "is_first_course": serializers.BooleanField(),
                    "is_faculty_active": serializers.BooleanField(),
                    "has_adapters": serializers.BooleanField(),
                    "is_done": serializers.BooleanField(),
                },
            ),
            status.HTTP_404_NOT_FOUND: inline_serializer(
                name="SurveyCheckUserNotFound",
                fields={"error": serializers.CharField()},
            ),
        },
    )
    def get(self, request: Request):
        user = request.user
        if not hasattr(user, "profile"):
            return Response({"error": "Profile does not exist!"}, status.HTTP_404_NOT_FOUND)

        is_first_course = user.profile.course and user.profile.course == 1
        is_faculty_active = (
            SurveyFaculty.objects.filter(name=user.profile.faculty).values_list("is_active", flat=True).first()
        )
        has_adapters = SurveyGroup.objects.filter(name=user.profile.group, adapters__isnull=False).exists()
        is_done = SurveyAnswer.objects.filter(user=user).exists()
        return Response(
            {
                "access": all((is_first_course, is_faculty_active, has_adapters, not is_done)),
                "is_first_course": is_first_course,
                "is_faculty_active": is_faculty_active,
                "has_adapters": has_adapters,
                "is_done": is_done,
            }
        )

Опрос должен быть доступен не всем, поэтому создадим эндпоинт, который проверит все условия для пользователя и вернёт ответ, чтобы фронтенд мог решить, что делать дальше.

Проверим курс пользователя, его факультет и статус активности, его учебную группу и наличие привязанных к ней адаптеров и факт прохождения опроса.

import logging

import requests
from django.conf import settings
from drf_spectacular.utils import extend_schema, inline_serializer
from rest_framework import status, serializers
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView

from .serializers import SurveyAnswerSerializer
from .models import SurveyQuestion, SurveyAnswer


logger = logging.getLogger("root")

BE_BASE_URL = settings.BE_BASE_URL


class CreateSurveyAnswersView(APIView):
    permission_classes = [IsAuthenticated]
    serializer_class = SurveyAnswerSerializer(many=True)

    @extend_schema(
        summary="Save user's answers and calculate scores",
        responses={
            status.HTTP_200_OK: inline_serializer(
                name="CreateSurveyAnswers",
                fields={
                    "status": serializers.CharField(),
                },
            ),
            status.HTTP_400_BAD_REQUEST: SurveyAnswerSerializer,
            status.HTTP_404_NOT_FOUND: inline_serializer(
                name="SurveyCheckUserProfileNotFound",
                fields={"error": serializers.CharField()},
            ),
            status.HTTP_409_CONFLICT: inline_serializer(
                name="SurveyCheckUserConflict",
                fields={"error": serializers.CharField()},
            ),
            status.HTTP_500_INTERNAL_SERVER_ERROR: inline_serializer(
                name="SurveyCheckUserServerError",
                fields={"error": serializers.CharField()},
            ),
        },
    )
    def post(self, request: Request):
        user = request.user
        if not hasattr(user, "profile"):
            return Response({"error": "Profile does not exist!"}, status.HTTP_404_NOT_FOUND)

        response = requests.get(f"{BE_BASE_URL}/api/survey/user/check/", headers=request.headers)
        response.raise_for_status()

        if not response.json().get("access", False):
            return Response({"error": "The survey checks has not been passed!"}, status=status.HTTP_409_CONFLICT)

        serializer = SurveyAnswerSerializer(data=request.data, many=True)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        answers = serializer.validated_data
        answers_to_create = []

        for answer in answers:
            question_id = answer["id"]
            value = answer["value"]

            question = SurveyQuestion.objects.filter(id=question_id).first()
            if not question:
                continue

            question_type = SurveyQuestion.Type(question.component)

            if question_type == SurveyQuestion.Type.TEXT_AREA:
                answers_to_create.append(SurveyAnswer(user=request.user, question=question, value=str(value), score=0))

            elif question_type in SurveyQuestion.Type.SELECT:
                answers_to_create.append(
                    SurveyAnswer(user=request.user, question=question, value=value, score=5 if value == "Да" else 0)
                )

            elif question_type in [SurveyQuestion.Type.SLIDER, SurveyQuestion.Type.TRAINING_SELECT]:
                answers_to_create.append(SurveyAnswer(user=request.user, question=question, value=value, score=value))

            elif question_type == SurveyQuestion.Type.ADAPTER_SLIDER:
                answers_to_create.extend(
                    SurveyAnswer(
                        user=request.user,
                        question=question,
                        adapter_id=adapter["id"],
                        value=str(adapter["value"]),
                        score=adapter["value"],
                    )
                    for adapter in value
                )

            else:
                continue

        try:
            SurveyAnswer.objects.bulk_create(answers_to_create)
        except Exception as exc:
            logger.exception(exc)
            return Response(
                {"error": "An error occurred while creating answers!"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

        return Response({"status": "success"})

В этом эндпоинте в POST запросе будем проверять отправленные ответы.

Для начала нам снова нужно проверить условия для прохождения опроса пользователем, для этого отправим запрос на эндпоинт проверки.

Если условия соблюдены, нужно проверить входные данные с помощью написанного ранее сериализатора для ответов.

Если все данные корректны, создадим список объектов SurveyAnswer, который будем создавать, проверяя различные условия, затем массово создадим все записи в БД с помощью метода bulk_create.