Add dandi exception classes

dandiapi/api/services/dandiset/__init__.py

@@ -1,8 +1,9 @@
-from django.core.exceptions import PermissionDenied, ValidationError
 from django.db import transaction
 
 from dandiapi.api.models.dandiset import Dandiset
 from dandiapi.api.models.version import Version
+from dandiapi.api.services.dandiset.exceptions import DandisetAlreadyExists
+from dandiapi.api.services.exceptions import AdminOnlyOperation, NotAllowed
 from dandiapi.api.services.version.metadata import _normalize_version_metadata
 
 
@@ -15,13 +16,13 @@ def create_dandiset(
     version_metadata: dict,
 ) -> tuple[Dandiset, Version]:
     if identifier and not user.is_superuser:
-        raise PermissionDenied(
+        raise AdminOnlyOperation(
             'Creating a dandiset for a given identifier is an admin only operation.'
         )
 
     existing_dandiset = Dandiset.objects.filter(id=identifier).first()
     if existing_dandiset:
-        raise ValidationError(f'Dandiset {existing_dandiset.identifier} already exists')
+        raise DandisetAlreadyExists(f'Dandiset {existing_dandiset.identifier} already exists')
 
     embargo_status = Dandiset.EmbargoStatus.EMBARGOED if embargo else Dandiset.EmbargoStatus.OPEN
     version_metadata = _normalize_version_metadata(
@@ -50,11 +51,11 @@ def delete_dandiset(*, user, dandiset: Dandiset) -> None:
         not user.has_perm('owner', dandiset)
         or dandiset.embargo_status != Dandiset.EmbargoStatus.OPEN
     ):
-        raise PermissionDenied()
+        raise NotAllowed()
 
     if dandiset.versions.exclude(version='draft').exists():
-        raise PermissionDenied('Cannot delete dandisets with published versions.')
+        raise NotAllowed('Cannot delete dandisets with published versions.')
     if dandiset.versions.filter(status=Version.Status.PUBLISHING).exists():
-        raise PermissionDenied('Cannot delete dandisets that are currently being published.')
+        raise NotAllowed('Cannot delete dandisets that are currently being published.')
 
     dandiset.delete()

dandiapi/api/services/dandiset/exceptions.py

@@ -0,0 +1,7 @@
+from rest_framework import status
+
+from dandiapi.api.services.exceptions import DandiException
+
+
+class DandisetAlreadyExists(DandiException):
+    http_status_code = status.HTTP_400_BAD_REQUEST

dandiapi/api/services/exceptions.py

@@ -0,0 +1,20 @@
+from rest_framework import status
+
+
+class DandiException(Exception):
+    def __init__(
+        self, message: str | None = None, http_status_code: int | None = None, *args: object
+    ) -> None:
+        self.message = message or self.message
+        self.http_status_code = http_status_code or self.http_status_code
+
+        super().__init__(*args)
+
+
+class NotAllowed(DandiException):
+    message = 'Action not allowed'
+    http_status_code = status.HTTP_403_FORBIDDEN
+
+
+class AdminOnlyOperation(DandiException):
+    pass

dandiapi/drf_utils.py

@@ -8,6 +8,8 @@
 from rest_framework.serializers import as_serializer_error
 from rest_framework.views import exception_handler
 
+from dandiapi.api.services.exceptions import DandiException
+
 
 def rewrap_django_core_exceptions(exc: Exception, ctx: dict) -> Response | None:
     """
@@ -23,6 +25,8 @@ def rewrap_django_core_exceptions(exc: Exception, ctx: dict) -> Response | None:
             return Response(exc.error_list[0].message, status=status.HTTP_400_BAD_REQUEST)
         else:
             exc = drf_exceptions.ValidationError(as_serializer_error(exc))
+    elif isinstance(exc, DandiException):
+        return Response(exc.message, status=exc.http_status_code or status.HTTP_400_BAD_REQUEST)
 
     if isinstance(exc, Http404):
         exc = drf_exceptions.NotFound()

tox.ini

@@ -87,6 +87,8 @@ ignore =
     D10,
     # variables should be lowercased
     N806,
+    # exceptions need Error in their name
+    N818
 extend-exclude =
     build,
     dist,