refactor!: replace django rest framework with django-ninja

dpdapi/api.py

@@ -0,0 +1,272 @@
+import logging
+import operator
+from datetime import datetime
+from functools import reduce
+from typing import Annotated
+
+from django.contrib.auth import authenticate
+from django.core.exceptions import ValidationError as DjangoValidationError
+from django.core.validators import validate_email
+from django.db import models, transaction
+from django.db.models import Q
+from django.shortcuts import get_object_or_404
+from django.utils import timezone
+from ninja import NinjaAPI, PatchDict, Query, Schema, Status
+from ninja.errors import ValidationError
+from ninja.security import HttpBasicAuth, django_auth
+from passlib.hash import sha512_crypt
+from pydantic import AfterValidator, PlainSerializer, StringConstraints
+
+from dpdapi.models import Alias, Domain, User
+
+logger = logging.getLogger(__name__)
+
+
+class BasicAuth(HttpBasicAuth):
+    def authenticate(self, request, username, password):
+        return authenticate(request, username=username, password=password)
+
+
+api = NinjaAPI(title="Postfix/Dovecot API", auth=[BasicAuth(), django_auth], urls_namespace="api")
+
+
+@api.exception_handler(ValidationError)
+def validation_error(request, exc):
+    return api.create_response(request, {"detail": exc.errors}, status=400)
+
+
+def _email(value: str) -> str:
+    try:
+        validate_email(value)
+    except DjangoValidationError as e:
+        raise ValueError("Enter a valid email address.") from e
+    return value.lower()
+
+
+Email = Annotated[str, StringConstraints(strip_whitespace=True, max_length=254), AfterValidator(_email)]
+Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=254)]
+Password = Annotated[str, StringConstraints(min_length=1, max_length=512)]
+# Rendered in the site's time zone, as Django REST framework did.
+LocalDateTime = Annotated[datetime, PlainSerializer(lambda v: timezone.localtime(v).isoformat())]
+
+
+def invalid(field: str, msg: str) -> ValidationError:
+    return ValidationError([{"loc": ["body", "payload", field], "msg": msg, "type": "value_error"}])
+
+
+def check_domains(domain_ids: list[int]):
+    """Reject domain ids that don't exist."""
+    if missing := set(domain_ids) - set(Domain.objects.filter(pk__in=domain_ids).values_list("pk", flat=True)):
+        raise invalid("domain", f"Invalid pk {sorted(missing)} - object does not exist.")
+
+
+def check_unique(instance: models.Model, field: str, value: str):
+    qs = type(instance).objects.filter(**{field: value}).exclude(pk=instance.pk)
+    if qs.exists():
+        raise invalid(field, f"{instance._meta.verbose_name} with this {field} already exists.")
+
+
+def not_null(data: dict) -> dict:
+    """PatchDict makes every field optional *and* nullable; only the former is wanted."""
+    for field, value in data.items():
+        if value is None:
+            raise invalid(field, "This field may not be null.")
+    return data
+
+
+class Timestamped(Schema):
+    id: int
+    created: LocalDateTime
+    updated: LocalDateTime
+
+
+# Aliases
+
+
+class AliasKey(Schema):
+    source: Email
+    destination: Email
+    domain: int
+
+
+class AliasIn(AliasKey):
+    active: bool = True
+
+
+class AliasOut(Timestamped):
+    source: str
+    destination: str
+    domain: int
+    active: bool
+
+    @staticmethod
+    def resolve_domain(obj):
+        return obj.domain_id
+
+
+class AliasFilters(Schema):
+    domain: int | None = None
+    domain__name: str | None = None
+    source__regex: str | None = None
+    source__iregex: str | None = None
+    destination__regex: str | None = None
+    destination__iregex: str | None = None
+
+
+class Result(Schema):
+    result: str
+
+
+@api.get("/aliases/", response=list[AliasOut], tags=["aliases"])
+def list_aliases(request, filters: Query[AliasFilters]):
+    return Alias.objects.filter(**filters.model_dump(exclude_none=True))
+
+
+@api.post("/aliases/create_bulk/", response={201: list[AliasOut]}, tags=["aliases"])
+@transaction.atomic
+def create_aliases(request, payload: list[AliasIn]):
+    """Create aliases, returning existing ones instead of failing on duplicates."""
+    check_domains([a.domain for a in payload])
+    aliases = [
+        Alias.objects.get_or_create(
+            source=a.source, destination=a.destination, domain_id=a.domain, defaults={"active": a.active}
+        )[0]
+        for a in payload
+    ]
+    return Status(201, aliases)
+
+
+@api.delete("/aliases/delete_bulk/", response={200: Result, 204: None}, tags=["aliases"])
+@transaction.atomic
+def delete_aliases(request, payload: list[AliasKey]):
+    """Delete aliases, reporting the ones that didn't exist."""
+    check_domains([a.domain for a in payload])
+    keys = {(a.source, a.destination, a.domain) for a in payload}
+    query = reduce(operator.or_, (Q(source=s, destination=d, domain_id=dom) for s, d, dom in keys), Q(pk__in=[]))
+    aliases = Alias.objects.filter(query)
+    deleted = set(aliases.values_list("source", "destination", "domain"))
+    aliases.delete()
+    logger.debug("Deleted: %s", ",".join(f"{s}={d}" for s, d, _ in sorted(deleted)))
+
+    if missing := keys - deleted:
+        formatted = ", ".join(f"{s}={d}" for s, d, _ in sorted(missing))
+        result = f"Aliases '{formatted}' could not be found and were not deleted."
+        logger.info(result)
+        return Status(200, {"result": result})
+    return Status(204, None)
+
+
+# Domains
+
+
+class DomainIn(Schema):
+    name: Name
+
+
+class DomainOut(Timestamped):
+    name: str
+
+
+class DomainFilters(Schema):
+    name: str | None = None
+
+
+def save_domain(domain: Domain, data: dict) -> Domain:
+    if "name" in data:
+        domain.name = data["name"].lower()
+        check_unique(domain, "name", domain.name)
+    domain.save()
+    return domain
+
+
+@api.get("/domains/", response=list[DomainOut], tags=["domains"])
+def list_domains(request, filters: Query[DomainFilters]):
+    return Domain.objects.filter(**filters.model_dump(exclude_none=True))
+
+
+@api.post("/domains/", response={201: DomainOut}, tags=["domains"])
+def create_domain(request, payload: DomainIn):
+    return Status(201, save_domain(Domain(), payload.model_dump()))
+
+
+@api.get("/domains/{int:pk}/", response=DomainOut, tags=["domains"])
+def get_domain(request, pk: int):
+    return get_object_or_404(Domain, pk=pk)
+
+
+@api.put("/domains/{int:pk}/", response=DomainOut, tags=["domains"])
+def update_domain(request, pk: int, payload: DomainIn):
+    return save_domain(get_object_or_404(Domain, pk=pk), payload.model_dump())
+
+
+@api.patch("/domains/{int:pk}/", response=DomainOut, tags=["domains"])
+def partial_update_domain(request, pk: int, payload: PatchDict[DomainIn]):
+    return save_domain(get_object_or_404(Domain, pk=pk), not_null(payload))
+
+
+@api.delete("/domains/{int:pk}/", response={204: None}, tags=["domains"])
+def delete_domain(request, pk: int):
+    get_object_or_404(Domain, pk=pk).delete()
+    return Status(204, None)
+
+
+# Users
+
+
+class UserIn(Schema):
+    email: Email
+    password: Password
+    domain: int
+
+
+class UserOut(Timestamped):
+    email: str
+    domain: int
+
+    @staticmethod
+    def resolve_domain(obj):
+        return obj.domain_id
+
+
+def save_user(user: User, data: dict) -> User:
+    if "email" in data:
+        user.email = data["email"]
+        check_unique(user, "email", user.email)
+    if "domain" in data:
+        check_domains([data["domain"]])
+        user.domain_id = data["domain"]
+    if "password" in data:
+        user.password = sha512_crypt.hash(data["password"])
+    user.save()
+    return user
+
+
+@api.get("/users/", response=list[UserOut], tags=["users"])
+def list_users(request):
+    return User.objects.all()
+
+
+@api.post("/users/", response={201: UserOut}, tags=["users"])
+def create_user(request, payload: UserIn):
+    return Status(201, save_user(User(), payload.model_dump()))
+
+
+@api.get("/users/{int:pk}/", response=UserOut, tags=["users"])
+def get_user(request, pk: int):
+    return get_object_or_404(User, pk=pk)
+
+
+@api.put("/users/{int:pk}/", response=UserOut, tags=["users"])
+def update_user(request, pk: int, payload: UserIn):
+    return save_user(get_object_or_404(User, pk=pk), payload.model_dump())
+
+
+@api.patch("/users/{int:pk}/", response=UserOut, tags=["users"])
+def partial_update_user(request, pk: int, payload: PatchDict[UserIn]):
+    return save_user(get_object_or_404(User, pk=pk), not_null(payload))
+
+
+@api.delete("/users/{int:pk}/", response={204: None}, tags=["users"])
+def delete_user(request, pk: int):
+    get_object_or_404(User, pk=pk).delete()
+    return Status(204, None)

dpdapi/filters.py

@@ -1,18 +0,0 @@
-import django_filters
-from rest_framework.filters import BaseFilterBackend
-
-from dpdapi.models import Alias
-
-
-class AliasRegexFilter(django_filters.FilterSet):
-    class Meta:
-        model = Alias
-        fields = {
-            'source': ['iregex', 'regex'],
-            'destination': ['iregex', 'regex']
-        }
-
-
-class AliasRegexFilterBackend(BaseFilterBackend):
-    def filter_queryset(self, request, queryset, view):
-        return AliasRegexFilter(request.query_params, queryset=queryset).qs

dpdapi/serializers.py

@@ -1,54 +0,0 @@
-from rest_framework import serializers
-from passlib.hash import sha512_crypt
-from .models import Alias, Domain, User
-
-
-class AliasSerializer(serializers.ModelSerializer):
-
-    def create(self, validated_data):
-        # Workaround to accept duplicates
-        # FIXME, will fail with existing alias with validated_data['active']=False
-        alias, created = Alias.objects.get_or_create(**validated_data)
-        return alias
-
-    class Meta:
-        model = Alias
-        validators = []  # Disable unique togheter check
-        fields = '__all__'
-
-
-class AliasDeleteSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = Alias
-        validators = []  # No Unique togheter check on deletion
-        fields = '__all__'
-
-
-class DomainSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = Domain
-        fields = '__all__'
-
-
-class UserSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = User
-        fields = '__all__'
-        extra_kwargs = {'password': {'write_only': True}}
-
-    def _hash_password(self, raw_password):
-        return sha512_crypt.hash(raw_password)
-
-    def create(self, validated_data):
-        user = User(
-            email=validated_data['email'],
-            domain=validated_data['domain'],
-            password=self._hash_password(validated_data['password'])
-        )
-        user.save()
-        return user
-
-    def update(self, instance, validated_data):
-        if 'password' in validated_data:
-            validated_data['password'] = self._hash_password(validated_data['password'])
-        return super().update(instance, validated_data)

dpdapi/urls.py

@@ -1,10 +0,0 @@
-from rest_framework.routers import DefaultRouter
-
-from .views import AliasViewSet, DomainViewSet, UserViewSet
-
-
-router = DefaultRouter()
-router.register(r'aliases', AliasViewSet)
-router.register(r'domains', DomainViewSet)
-router.register(r'users', UserViewSet)
-urlpatterns = router.urls

dpdapi/views.py

@@ -1,75 +0,0 @@
-import logging
-import operator
-
-from functools import reduce
-from django.db.models import Q
-from django_filters.rest_framework import DjangoFilterBackend
-from rest_framework.decorators import action
-from rest_framework.mixins import ListModelMixin
-from rest_framework.permissions import IsAuthenticated
-from rest_framework.response import Response
-from rest_framework import status, viewsets
-from rest_framework.viewsets import GenericViewSet
-
-from dpdapi.filters import AliasRegexFilterBackend
-from dpdapi.models import Alias, Domain, User
-from dpdapi.serializers import AliasSerializer, AliasDeleteSerializer, DomainSerializer, UserSerializer
-
-logger = logging.getLogger(__name__)
-
-
-class AliasViewSet(ListModelMixin, GenericViewSet):
-    """ Aliases """
-    queryset = Alias.objects.all()
-    serializer_class = AliasSerializer
-    permission_classes = [IsAuthenticated]
-    filter_backends = (DjangoFilterBackend, AliasRegexFilterBackend)
-    filterset_fields = ('domain', 'domain__name',)
-
-    @action(detail=False, methods=['post'])
-    def create_bulk(self, request):
-        serializer = AliasSerializer(data=request.data, many=True)
-
-        serializer.is_valid(raise_exception=True)
-
-        serializer.save()
-        return Response(serializer.data, status=status.HTTP_201_CREATED)
-
-    @action(detail=False, methods=['delete'])
-    def delete_bulk(self, request):
-        serializer = AliasDeleteSerializer(data=request.data, many=True)
-
-        serializer.is_valid(raise_exception=True)
-
-        keys = {(a['source'], a['destination'], a['domain'].pk) for a in serializer.validated_data}
-        query = reduce(operator.or_, (Q(source=s, destination=d, domain=dom) for s, d, dom in keys), Q(pk__in=[]))
-        aliases = Alias.objects.filter(query)
-
-        deleted_set = set(aliases.values_list('source', 'destination', 'domain'))
-        diff = keys - deleted_set
-        logger.debug('Deleted: {}'.format(','.join(map(lambda x: '{0[0]}={0[1]}'.format(x), deleted_set))))
-        aliases.delete()
-
-        if diff:
-            diff_formatted = map(lambda x: '{0[0]}={0[1]}'.format(x), diff)
-            res = "Aliases '{}' could not be found and were not deleted.".format(", ".join(diff_formatted))
-            logger.info(res)
-            return Response({"result": res}, status=status.HTTP_200_OK)
-
-        return Response(status=status.HTTP_204_NO_CONTENT)
-
-
-class DomainViewSet(viewsets.ModelViewSet):
-    """ Domains """
-    queryset = Domain.objects.all()
-    serializer_class = DomainSerializer
-    permission_classes = [IsAuthenticated]
-    filter_backends = (DjangoFilterBackend,)
-    filterset_fields = ('name',)
-
-
-class UserViewSet(viewsets.ModelViewSet):
-    """ Users """
-    queryset = User.objects.all()
-    serializer_class = UserSerializer
-    permission_classes = [IsAuthenticated]

dpdapisite/urls.py

@@ -1,10 +1,11 @@
 from django.contrib import admin
-from django.urls import include, path
+from django.urls import path
 from django.views.generic import RedirectView
 
+from dpdapi.api import api
+
 urlpatterns = [
-    path("", RedirectView.as_view(pattern_name="api-root")),
+    path("", RedirectView.as_view(pattern_name="api:openapi-view")),
-    path("api/", include("dpdapi.urls")),
+    path("api/", api.urls),
     path("admin/", admin.site.urls),
-    path("api-auth/", include("rest_framework.urls", namespace="rest_framework")),
 ]