from collections import namedtuple from rest_framework import generics from rest_framework import status from rest_framework.response import Response from translation import models as translation_models from utils import exceptions from rest_framework_simplejwt import tokens from django.conf import settings # JWT # Login base view mixin class JWTGenericViewMixin(generics.GenericAPIView): """JWT view mixin""" LOCALE_HTTP_ONLY = False LOCALE_SECURE = False ACCESS_TOKEN_HTTP_ONLY = False ACCESS_TOKEN_SECURE = False REFRESH_TOKEN_HTTP_ONLY = False REFRESH_TOKEN_SECURE = False COOKIE = namedtuple('COOKIE', ['key', 'value', 'http_only', 'secure', 'max_age']) def _create_jwt_token(self, user) -> dict: """Return dictionary with pairs access and refresh tokens""" token = tokens.RefreshToken.for_user(user) token['user'] = user.get_user_info() return { 'access_token': str(token.access_token), 'refresh_token': str(token), } def _get_locale(self, request): """Get locale from request""" return request.COOKIES.get('locale') def _check_locale(self, locale: str): locale_qs = translation_models.Language.objects.by_locale(locale=locale) if not locale_qs.exists(): raise exceptions.LocaleNotExisted() return locale def _put_data_in_cookies(self, locale: str, access_token: str = None, refresh_token: str = None, permanent: bool = None): """ CHECK locale in cookies and PUT access and refresh tokens there. cookies it is list that contain namedtuples cookies would contain key, value and secure parameters. """ COOKIES = list() # Create locale namedtuple _locale = self.COOKIE(key='locale', value=locale, http_only=self.LOCALE_HTTP_ONLY, secure=self.LOCALE_SECURE, max_age=None if permanent else settings.COOKIES_MAX_AGE) # Write to cookie access and refresh token with secure flag if access_token and refresh_token: _access_token = self.COOKIE(key='access_token', value=access_token, http_only=self.ACCESS_TOKEN_HTTP_ONLY, secure=self.ACCESS_TOKEN_SECURE, max_age=None if permanent else settings.COOKIES_MAX_AGE) _refresh_token = self.COOKIE(key='refresh_token', value=refresh_token, http_only=self.REFRESH_TOKEN_HTTP_ONLY, secure=self.REFRESH_TOKEN_SECURE, max_age=None if permanent else settings.COOKIES_MAX_AGE) COOKIES.extend((_access_token, _refresh_token)) COOKIES.append(_locale) return COOKIES def _put_cookies_in_response(self, cookies: list, response: Response): """Update COOKIES in response from namedtuple""" for cookie in cookies: # todo: remove config for develop import os configuration = os.environ.get('SETTINGS_CONFIGURATION', None) if configuration == 'development': response.set_cookie(key=cookie.key, value=cookie.value, secure=cookie.secure, httponly=cookie.http_only, max_age=cookie.max_age, domain='.id-east.ru') else: response.set_cookie(key=cookie.key, value=cookie.value, secure=cookie.secure, httponly=cookie.http_only, max_age=cookie.max_age,) return response def _get_tokens_from_cookies(self, request, cookies: dict = None): """Get user tokens from cookies and put in namedtuple""" _cookies = request.COOKIES or cookies return [self.COOKIE(key='access_token', value=_cookies.get('access_token'), http_only=self.ACCESS_TOKEN_HTTP_ONLY, secure=self.ACCESS_TOKEN_SECURE), self.COOKIE(key='refresh_token', value=_cookies.get('refresh_token'), http_only=self.REFRESH_TOKEN_HTTP_ONLY, secure=self.REFRESH_TOKEN_SECURE)] class JWTListAPIView(JWTGenericViewMixin, generics.ListAPIView): """ Concrete view for creating a model instance. """ def get(self, request, *args, **kwargs): _locale = self._get_locale(request=request) try: locale = self._check_locale(locale=_locale) queryset = self.filter_queryset(self.get_queryset()) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) response = self.get_paginated_response(serializer.data) else: serializer = self.get_serializer(queryset, many=True) response = Response(serializer.data) access_token, refresh_token = self._get_tokens_from_cookies(request) except exceptions.LocaleNotExisted: raise exceptions.LocaleNotExisted(locale=_locale) else: return self._put_cookies_in_response( cookies=self._put_data_in_cookies(locale=locale, access_token=access_token.value, refresh_token=refresh_token.value), response=response) class JWTCreateAPIView(JWTGenericViewMixin, generics.CreateAPIView): """ Concrete view for creating a model instance. """ def post(self, request, *args, **kwargs): _locale = self._get_locale(request=request) try: locale = self._check_locale(locale=_locale) serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) serializer.save() response = Response(serializer.data, status=status.HTTP_201_CREATED) access_token, refresh_token = self._get_tokens_from_cookies(request) except exceptions.LocaleNotExisted: raise exceptions.LocaleNotExisted(locale=_locale) else: return self._put_cookies_in_response( cookies=self._put_data_in_cookies(locale=locale, access_token=access_token.value, refresh_token=refresh_token.value), response=response) class JWTRetrieveAPIView(JWTGenericViewMixin, generics.RetrieveAPIView): """ Concrete view for retrieving a model instance. """ def get(self, request, *args, **kwargs): """Implement GET method""" _locale = self._get_locale(request=request) try: locale = self._check_locale(locale=_locale) queryset = self.filter_queryset(self.get_queryset()) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) response = self.get_paginated_response(serializer.data) else: serializer = self.get_serializer(queryset, many=True) response = Response(serializer.data, status.HTTP_200_OK) access_token, refresh_token = self._get_tokens_from_cookies(request) except exceptions.LocaleNotExisted: raise exceptions.LocaleNotExisted(locale=_locale) else: return self._put_cookies_in_response( cookies=self._put_data_in_cookies(locale=locale, access_token=access_token, refresh_token=refresh_token), response=response) class JWTDestroyAPIView(JWTGenericViewMixin, generics.DestroyAPIView): """ Concrete view for deleting a model instance. """ def delete(self, request, *args, **kwargs): _locale = self._get_locale(request=request) try: locale = self._check_locale(locale=_locale) instance = self.get_object() instance.delete() response = Response(status=status.HTTP_204_NO_CONTENT) access_token, refresh_token = self._get_tokens_from_cookies(request) except exceptions.LocaleNotExisted: raise exceptions.LocaleNotExisted(locale=_locale) else: return self._put_cookies_in_response( cookies=self._put_data_in_cookies(locale=locale, access_token=access_token, refresh_token=refresh_token), response=response) class JWTUpdateAPIView(JWTGenericViewMixin, generics.UpdateAPIView): """ Concrete view for updating a model instance. """ def put(self, request, *args, **kwargs): _locale = self._get_locale(request=request) try: locale = self._check_locale(locale=_locale) 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) serializer.save() 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 = {} response = Response(serializer.data) access_token, refresh_token = self._get_tokens_from_cookies(request) except exceptions.LocaleNotExisted: raise exceptions.LocaleNotExisted(locale=_locale) else: return self._put_cookies_in_response( cookies=self._put_data_in_cookies(locale=locale, access_token=access_token, refresh_token=refresh_token), response=response) def patch(self, request, *args, **kwargs): kwargs['partial'] = True return self.put(request, *args, **kwargs)