93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from drf_spectacular.utils import (
|
|
OpenApiParameter,
|
|
OpenApiResponse,
|
|
extend_schema,
|
|
)
|
|
from rest_framework import status
|
|
from rest_framework.response import Response
|
|
|
|
from ..base import SessionBaseAPIView
|
|
from ..models import UserAuthSession
|
|
from ..serializers import (
|
|
SessionBulkActionResponseSerializer,
|
|
SessionSerializer,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SessionListAPIView(SessionBaseAPIView):
|
|
@extend_schema(
|
|
tags=["Auth"],
|
|
summary="Список активных сессий пользователя",
|
|
responses={200: SessionSerializer(many=True)},
|
|
)
|
|
def get(self, request):
|
|
current_session_id = self.get_current_session_id(request)
|
|
queryset = UserAuthSession.objects.filter(
|
|
user=request.user,
|
|
revoked_at__isnull=True,
|
|
).order_by("-last_seen_at", "-created_at")
|
|
serializer = SessionSerializer(
|
|
queryset,
|
|
many=True,
|
|
context={"current_session_id": current_session_id},
|
|
)
|
|
return Response(serializer.data)
|
|
|
|
|
|
class SessionDetailAPIView(SessionBaseAPIView):
|
|
@extend_schema(
|
|
tags=["Auth"],
|
|
summary="Завершить одну сессию",
|
|
parameters=[
|
|
OpenApiParameter(
|
|
name="session_id",
|
|
type=str,
|
|
location=OpenApiParameter.PATH,
|
|
)
|
|
],
|
|
responses={204: OpenApiResponse(description="Сессия завершена.")},
|
|
)
|
|
def delete(self, request, session_id):
|
|
session = UserAuthSession.objects.filter(
|
|
user=request.user,
|
|
session_id=session_id,
|
|
revoked_at__isnull=True,
|
|
).first()
|
|
if not session:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
reason = (
|
|
"self_logout"
|
|
if str(session.session_id) == str(self.get_current_session_id(request))
|
|
else "manual_logout"
|
|
)
|
|
session.revoke(reason=reason)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
class LogoutOtherSessionsAPIView(SessionBaseAPIView):
|
|
@extend_schema(
|
|
tags=["Auth"],
|
|
summary="Завершить все остальные сессии",
|
|
responses={200: SessionBulkActionResponseSerializer},
|
|
)
|
|
def post(self, request):
|
|
current_session_id = self.get_current_session_id(request)
|
|
sessions = UserAuthSession.objects.filter(
|
|
user=request.user,
|
|
revoked_at__isnull=True,
|
|
).exclude(session_id=current_session_id)
|
|
|
|
revoked = 0
|
|
for session in sessions:
|
|
session.revoke(reason="logout_others")
|
|
revoked += 1
|
|
|
|
return Response({"revoked": revoked}, status=status.HTTP_200_OK)
|