authentik.api.validation

 1from collections.abc import Callable
 2from functools import wraps
 3from typing import Literal
 4
 5from rest_framework.request import Request
 6from rest_framework.response import Response
 7from rest_framework.serializers import Serializer
 8from rest_framework.viewsets import ViewSet
 9
10
11def validate(serializer_type: type[Serializer], location: Literal["body", "query"] = "body"):
12    """Validate incoming data with the specified serializer. Raw data can either be taken
13    from request body or query string, defaulting to body.
14
15    Validated data is added to the function this decorator is used on with a named parameter
16    based on the location of the data.
17
18    Example:
19
20        @validate(MySerializer)
21        @validate(MyQuerySerializer, location="query")
22        def my_action(self, request, *, body: MySerializer, query: MyQuerySerializer):
23            ...
24
25    """
26
27    def wrapper_outer(func: Callable):
28
29        @wraps(func)
30        def wrapper(self: ViewSet, request: Request, *args, **kwargs) -> Response:
31            data = {}
32            if location == "body":
33                data = request.data
34            elif location == "query":
35                data = request.query_params
36            else:
37                raise ValueError(f"Invalid data location '{location}'")
38            instance = serializer_type(
39                data=data,
40                context={
41                    "request": request,
42                },
43            )
44            instance.is_valid(raise_exception=True)
45            kwargs[location] = instance
46            return func(self, request, *args, **kwargs)
47
48        return wrapper
49
50    return wrapper_outer