Egor fureunoir Gorbunov
a81f734e23
Fixes: (1) Removed all `# type: ignore` annotations across the codebase; (2) Fixed usage of Django Model methods by eliminating unnecessary `# type: ignore` directives; (3) Adjusted usage of functions like `get()` to align with method expectations, removing incorrect comments; Extra: (1) Deleted `pyrightconfig.json` as part of migration to a stricter type-checked environment; (2) Minor code cleanup, including formatting changes and refactoring import statements in adherence to PEP8 recommendations.
105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
import logging
|
|
from typing import Any, Type
|
|
|
|
from django.utils.decorators import method_decorator
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django_ratelimit.decorators import ratelimit
|
|
from drf_spectacular.utils import (
|
|
extend_schema_view,
|
|
)
|
|
from rest_framework import status
|
|
from rest_framework.permissions import AllowAny, BasePermission
|
|
from rest_framework.request import Request
|
|
from rest_framework.response import Response
|
|
from rest_framework_simplejwt.exceptions import TokenError
|
|
from rest_framework_simplejwt.views import TokenViewBase
|
|
|
|
from engine.vibes_auth.docs.drf.views import (
|
|
TOKEN_OBTAIN_SCHEMA,
|
|
TOKEN_REFRESH_SCHEMA,
|
|
TOKEN_VERIFY_SCHEMA,
|
|
)
|
|
from engine.vibes_auth.serializers import (
|
|
TokenObtainPairSerializer,
|
|
TokenRefreshSerializer,
|
|
TokenVerifySerializer,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@extend_schema_view(**TOKEN_OBTAIN_SCHEMA)
|
|
class TokenObtainPairView(TokenViewBase):
|
|
__doc__ = _(
|
|
"Represents a view for getting a pair of access and refresh tokens and user's data. "
|
|
"This view manages the process of handling token-based authentication where "
|
|
"clients can get a pair of JWT tokens (access and refresh) using provided "
|
|
"credentials. It is built on top of a base token view and ensures proper "
|
|
"rate limiting to protect against brute force attacks."
|
|
)
|
|
|
|
serializer_class = TokenObtainPairSerializer
|
|
_serializer_class = TokenObtainPairSerializer
|
|
permission_classes: list[Type[BasePermission]] = [
|
|
AllowAny,
|
|
]
|
|
authentication_classes: list[str] = []
|
|
|
|
@method_decorator(ratelimit(key="ip", rate="10/h"))
|
|
def post(
|
|
self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]
|
|
) -> Response:
|
|
return super().post(request, *args, **kwargs)
|
|
|
|
|
|
@extend_schema_view(**TOKEN_REFRESH_SCHEMA)
|
|
class TokenRefreshView(TokenViewBase):
|
|
__doc__ = _(
|
|
"Handles refreshing of tokens for authentication purposes. "
|
|
"This class is used to provide functionality for token refresh "
|
|
"operations as part of an authentication system. It ensures that "
|
|
"clients can request a refreshed token within defined rate limits. "
|
|
"The view relies on the associated serializer to validate token "
|
|
"refresh inputs and produce appropriate outputs."
|
|
)
|
|
|
|
serializer_class = TokenRefreshSerializer
|
|
_serializer_class = TokenRefreshSerializer
|
|
permission_classes: list[Type[BasePermission]] = [
|
|
AllowAny,
|
|
]
|
|
authentication_classes: list[str] = []
|
|
|
|
@method_decorator(ratelimit(key="ip", rate="10/h"))
|
|
def post(
|
|
self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]
|
|
) -> Response:
|
|
return super().post(request, *args, **kwargs)
|
|
|
|
|
|
@extend_schema_view(**TOKEN_VERIFY_SCHEMA)
|
|
class TokenVerifyView(TokenViewBase):
|
|
__doc__ = _(
|
|
"Represents a view for verifying JSON Web Tokens (JWT) using specific serialization and validation logic. "
|
|
)
|
|
|
|
serializer_class = TokenVerifySerializer
|
|
_serializer_class = TokenVerifySerializer
|
|
permission_classes: list[Type[BasePermission]] = [
|
|
AllowAny,
|
|
]
|
|
authentication_classes: list[str] = []
|
|
|
|
def post(
|
|
self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]
|
|
) -> Response:
|
|
try:
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
user_data = serializer.validated_data.pop("user", None)
|
|
return Response({"token": _("the token is valid"), "user": user_data})
|
|
except TokenError:
|
|
return Response(
|
|
{"detail": _("the token is invalid")},
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|