118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
from django.contrib.auth import authenticate, login, logout
|
|
from django.middleware.csrf import get_token
|
|
from django.views.decorators.csrf import ensure_csrf_cookie
|
|
from django.utils.decorators import method_decorator
|
|
from rest_framework import permissions, status, viewsets
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from apps.accounts.models import ApiKey, User
|
|
from apps.accounts.serializers import (
|
|
ApiKeyCreateSerializer,
|
|
ApiKeySerializer,
|
|
LoginSerializer,
|
|
UserSerializer,
|
|
)
|
|
|
|
|
|
class IsOwnerOrStaff(permissions.BasePermission):
|
|
def has_object_permission(self, request, view, obj):
|
|
if request.user.is_staff or request.user.is_superuser:
|
|
return True
|
|
return getattr(obj, "user_id", None) == request.user.pk or obj == request.user
|
|
|
|
|
|
class IsAdminUser(permissions.IsAdminUser):
|
|
"""Admin access for staff OR superadmin (superadmins may have is_staff=False)."""
|
|
|
|
def has_permission(self, request, view):
|
|
return bool(request.user and (request.user.is_staff or request.user.is_superuser))
|
|
|
|
|
|
class UserViewSet(viewsets.ModelViewSet):
|
|
queryset = User.objects.all().order_by("user_name")
|
|
serializer_class = UserSerializer
|
|
permission_classes = [IsAdminUser]
|
|
|
|
|
|
class ApiKeyViewSet(viewsets.ModelViewSet):
|
|
http_method_names = ["get", "post", "delete", "head", "options"]
|
|
|
|
def get_queryset(self):
|
|
qs = ApiKey.objects.select_related("user").order_by("-created_at")
|
|
if self.request.user.is_staff or self.request.user.is_superuser:
|
|
return qs
|
|
return qs.filter(user=self.request.user)
|
|
|
|
def get_serializer_class(self):
|
|
if self.action == "create":
|
|
return ApiKeyCreateSerializer
|
|
return ApiKeySerializer
|
|
|
|
def get_permissions(self):
|
|
return [permissions.IsAuthenticated()]
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
name = serializer.validated_data["name"]
|
|
user = request.user
|
|
if (request.user.is_staff or request.user.is_superuser) and serializer.validated_data.get(
|
|
"user_id"
|
|
):
|
|
user = User.objects.get(pk=serializer.validated_data["user_id"])
|
|
api_key, raw = ApiKey.generate(user=user, name=name)
|
|
data = ApiKeySerializer(api_key).data
|
|
data["raw_key"] = raw
|
|
return Response(data, status=status.HTTP_201_CREATED)
|
|
|
|
def perform_destroy(self, instance):
|
|
if (
|
|
not self.request.user.is_staff
|
|
and not self.request.user.is_superuser
|
|
and instance.user_id != self.request.user.pk
|
|
):
|
|
raise permissions.PermissionDenied()
|
|
instance.delete()
|
|
|
|
|
|
@method_decorator(ensure_csrf_cookie, name="dispatch")
|
|
class CsrfView(APIView):
|
|
permission_classes = [permissions.AllowAny]
|
|
authentication_classes = []
|
|
|
|
def get(self, request):
|
|
return Response({"csrfToken": get_token(request)})
|
|
|
|
|
|
class LoginView(APIView):
|
|
permission_classes = [permissions.AllowAny]
|
|
authentication_classes = []
|
|
|
|
def post(self, request):
|
|
serializer = LoginSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
user = authenticate(
|
|
request,
|
|
username=serializer.validated_data["user_name"],
|
|
password=serializer.validated_data["password"],
|
|
)
|
|
if user is None:
|
|
return Response(
|
|
{"detail": "Invalid credentials"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
login(request, user)
|
|
return Response(UserSerializer(user).data)
|
|
|
|
|
|
class LogoutView(APIView):
|
|
def post(self, request):
|
|
logout(request)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
class MeView(APIView):
|
|
def get(self, request):
|
|
return Response(UserSerializer(request.user).data)
|