121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
from typing import Any
|
|
|
|
from rest_framework import status
|
|
from rest_framework.decorators import api_view, permission_classes, throttle_classes
|
|
from rest_framework.permissions import AllowAny
|
|
from rest_framework.request import Request
|
|
from rest_framework.response import Response
|
|
from rest_framework.throttling import AnonRateThrottle
|
|
from rest_framework_simplejwt.exceptions import TokenError
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
from accounts.serializers import (
|
|
LoginSerializer,
|
|
LogoutSerializer,
|
|
RefreshRequestSerializer,
|
|
RegisterSerializer,
|
|
UserSerializer,
|
|
)
|
|
|
|
|
|
class AuthRateThrottle(AnonRateThrottle):
|
|
scope = "auth"
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
@throttle_classes([AuthRateThrottle])
|
|
def register_view(request: Request) -> Response:
|
|
"""Register a new user account."""
|
|
serializer = RegisterSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
user = serializer.save()
|
|
return Response(
|
|
UserSerializer(user).data,
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
@throttle_classes([AuthRateThrottle])
|
|
def login_view(request: Request) -> Response:
|
|
"""Authenticate a user and return JWT tokens."""
|
|
serializer = LoginSerializer(
|
|
data=request.data,
|
|
context={"request": request},
|
|
)
|
|
serializer.is_valid(raise_exception=True)
|
|
validated_data: dict[str, Any] = serializer.validated_data
|
|
return Response(
|
|
{
|
|
"user": UserSerializer(validated_data["user"]).data,
|
|
"access": validated_data["access"],
|
|
"refresh": validated_data["refresh"],
|
|
},
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
@throttle_classes([AuthRateThrottle])
|
|
def token_refresh_view(request: Request) -> Response:
|
|
"""Refresh an access token using a refresh token.
|
|
|
|
Uses SimpleJWT's built-in rotation and blacklisting.
|
|
"""
|
|
serializer = RefreshRequestSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
refresh_token_str: str = serializer.validated_data["refresh"]
|
|
try:
|
|
refresh = RefreshToken(refresh_token_str)
|
|
access = str(refresh.access_token)
|
|
new_refresh = str(refresh)
|
|
except TokenError as e:
|
|
return Response(
|
|
{"error": str(e), "code": "token_invalid"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
|
|
return Response(
|
|
{"access": access, "refresh": new_refresh},
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
|
|
|
|
@api_view(["POST"])
|
|
@permission_classes([AllowAny])
|
|
def logout_view(request: Request) -> Response:
|
|
"""Blacklist a refresh token (log out).
|
|
|
|
This allows explicit token revocation on logout.
|
|
"""
|
|
serializer = LogoutSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
refresh_token_str: str = serializer.validated_data["refresh"]
|
|
try:
|
|
refresh = RefreshToken(refresh_token_str)
|
|
refresh.blacklist()
|
|
except TokenError:
|
|
# If token is already invalid/blacklisted, still consider logout successful
|
|
pass
|
|
except AttributeError:
|
|
# If blacklist app not installed
|
|
pass
|
|
|
|
return Response(
|
|
{"message": "Successfully logged out."},
|
|
status=status.HTTP_205_RESET_CONTENT,
|
|
)
|
|
|
|
|
|
@api_view(["GET"])
|
|
def me_view(request: Request) -> Response:
|
|
"""Return the current authenticated user's profile."""
|
|
return Response(
|
|
UserSerializer(request.user).data,
|
|
status=status.HTTP_200_OK,
|
|
) |