django-postfix-dovecot-api/dpdapi/api.py

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)