init

services/backend/api/v1/handlers.py

@@ -0,0 +1,121 @@
+import logging
+from collections.abc import Callable
+from http import HTTPStatus as status
+from typing import Any
+
+import django.core.exceptions
+import django.http
+import ninja.errors
+from django.http import HttpRequest, HttpResponse
+from ninja import NinjaAPI
+
+from config.errors import ConflictError, ForbiddenError
+
+logger = logging.getLogger("django")
+
+
+def handle_validation_error(
+    request: HttpRequest,
+    exc: ninja.errors.ValidationError,
+    router: NinjaAPI,
+) -> HttpResponse:
+    return router.create_response(
+        request,
+        {"detail": exc.errors},
+        status=status.BAD_REQUEST,
+    )
+
+
+def handle_django_validation_error(
+    request: HttpRequest,
+    exc: django.core.exceptions.ValidationError,
+    router: NinjaAPI,
+) -> HttpResponse:
+    detail = list(exc)
+
+    if hasattr(exc, "error_dict"):
+        detail = dict(exc)
+
+    return router.create_response(
+        request,
+        {"detail": detail},
+        status=status.BAD_REQUEST,
+    )
+
+
+def handle_authentication_error(
+    request: HttpRequest,
+    exc: ninja.errors.AuthenticationError,
+    router: NinjaAPI,
+) -> HttpResponse:
+    return router.create_response(
+        request,
+        {"detail": status.UNAUTHORIZED.phrase},
+        status=status.UNAUTHORIZED,
+    )
+
+
+def handle_forbidden_error(
+    request: HttpRequest,
+    exc: ForbiddenError,
+    router: NinjaAPI,
+) -> HttpResponse:
+    return router.create_response(
+        request,
+        {"detail": exc.message},
+        status=status.FORBIDDEN,
+    )
+
+
+def handle_not_found_error(
+    request: HttpRequest,
+    exc: Exception,
+    router: NinjaAPI,
+) -> HttpResponse:
+    return router.create_response(
+        request,
+        {"detail": status.NOT_FOUND.phrase},
+        status=status.NOT_FOUND,
+    )
+
+
+def handle_conflict_error(
+    request: HttpRequest,
+    exc: ConflictError,
+    router: NinjaAPI,
+) -> HttpResponse:
+    detail = list(exc.validation_error)
+
+    if hasattr(exc, "error_dict"):
+        detail = dict(exc.validation_error)
+
+    return router.create_response(
+        request,
+        {"detail": detail},
+        status=status.CONFLICT,
+    )
+
+
+def handle_unknown_exception(
+    request: HttpRequest,
+    exc: Exception,
+    router: NinjaAPI,
+) -> HttpResponse:
+    logger.exception(exc)
+
+    return router.create_response(
+        request,
+        {"detail": status.INTERNAL_SERVER_ERROR.phrase},
+        status=status.INTERNAL_SERVER_ERROR,
+    )
+
+
+exception_handlers: list[tuple[Any, Callable]] = [
+    (ninja.errors.ValidationError, handle_validation_error),
+    (django.core.exceptions.ValidationError, handle_django_validation_error),
+    (ninja.errors.AuthenticationError, handle_authentication_error),
+    (ForbiddenError, handle_forbidden_error),
+    (django.http.Http404, handle_not_found_error),
+    (ConflictError, handle_conflict_error),
+    (Exception, handle_unknown_exception),
+]

services/backend/api/v1/ping/schemas.py

@@ -0,0 +1,5 @@
+from ninja import Schema
+
+
+class PingOut(Schema):
+    status: str = "ok"

services/backend/api/v1/ping/views.py

@@ -0,0 +1,18 @@
+from http import HTTPStatus as status
+
+from django.http import HttpRequest
+from ninja import Router
+
+from api.v1.ping import schemas
+
+router = Router(tags=["ping"])
+
+
+@router.get(
+    "",
+    response={
+        status.OK: schemas.PingOut,
+    },
+)
+def ping(request: HttpRequest) -> tuple[status, schemas.PingOut]:
+    return status.OK, schemas.PingOut

services/backend/api/v1/router.py

@@ -0,0 +1,23 @@
+from functools import partial
+
+from ninja import NinjaAPI
+
+from api.v1 import handlers
+from api.v1.ping.views import router as ping_router
+
+router = NinjaAPI(
+    title="DataRush API",
+    version="1",
+    description="API docs for DataRush",
+    openapi_url="/docs/openapi.json",
+)
+
+
+router.add_router(
+    "ping",
+    ping_router,
+)
+
+
+for exception, handler in handlers.exception_handlers:
+    router.add_exception_handler(exception, partial(handler, router=router))

services/backend/api/v1/schemas.py

@@ -0,0 +1,24 @@
+from http import HTTPStatus as status
+from typing import Any
+
+from ninja import Schema
+
+
+class BadRequestError(Schema):
+    detail: Any
+
+
+class UnauthorizedError(Schema):
+    detail: str = status.UNAUTHORIZED.phrase
+
+
+class ForbiddenError(Schema):
+    detail: str = status.FORBIDDEN.phrase
+
+
+class NotFoundError(Schema):
+    detail: str = status.NOT_FOUND.phrase
+
+
+class ConflictError(Schema):
+    detail: Any