authentik.tasks.schedules.api 

  1from django_filters.filters import BooleanFilter
  2from django_filters.filterset import FilterSet
  3from dramatiq.actor import Actor
  4from dramatiq.broker import get_broker
  5from dramatiq.errors import ActorNotFound
  6from drf_spectacular.types import OpenApiTypes
  7from drf_spectacular.utils import OpenApiResponse, extend_schema
  8from rest_framework.decorators import action
  9from rest_framework.fields import ReadOnlyField
 10from rest_framework.mixins import (
 11    ListModelMixin,
 12    RetrieveModelMixin,
 13    UpdateModelMixin,
 14)
 15from rest_framework.request import Request
 16from rest_framework.response import Response
 17from rest_framework.serializers import SerializerMethodField
 18from rest_framework.viewsets import GenericViewSet
 19from structlog.stdlib import get_logger
 20
 21from authentik.core.api.utils import ModelSerializer
 22from authentik.rbac.decorators import permission_required
 23from authentik.tasks.models import Task, TaskStatus
 24from authentik.tasks.schedules.models import Schedule
 25
 26LOGGER = get_logger()
 27
 28
 29class ScheduleSerializer(ModelSerializer):
 30    rel_obj_app_label = ReadOnlyField(source="rel_obj_content_type.app_label")
 31    rel_obj_model = ReadOnlyField(source="rel_obj_content_type.model")
 32
 33    description = SerializerMethodField()
 34    last_task_status = SerializerMethodField()
 35
 36    class Meta:
 37        model = Schedule
 38        fields = (
 39            "id",
 40            "identifier",
 41            "uid",
 42            "actor_name",
 43            "rel_obj_app_label",
 44            "rel_obj_model",
 45            "rel_obj_id",
 46            "crontab",
 47            "paused",
 48            "next_run",
 49            "description",
 50            "last_task_status",
 51        )
 52
 53    def get_description(self, instance: Schedule) -> str | None:
 54        try:
 55            actor: Actor = get_broker().get_actor(instance.actor_name)
 56        except ActorNotFound:
 57            LOGGER.warning("Could not find actor for schedule", schedule=instance)
 58            return None
 59        if "description" not in actor.options:
 60            LOGGER.warning(
 61                "Could not find description for actor",
 62                schedule=instance,
 63                actor=actor.actor_name,
 64            )
 65            return None
 66        return actor.options["description"]
 67
 68    def get_last_task_status(self, instance: Schedule) -> TaskStatus | None:
 69        last_task: Task = instance.tasks.defer("message", "result").order_by("-mtime").first()
 70        if last_task:
 71            return last_task.aggregated_status
 72        return None
 73
 74
 75class ScheduleFilter(FilterSet):
 76    rel_obj_id__isnull = BooleanFilter("rel_obj_id", "isnull")
 77
 78    class Meta:
 79        model = Schedule
 80        fields = (
 81            "actor_name",
 82            "rel_obj_content_type__app_label",
 83            "rel_obj_content_type__model",
 84            "rel_obj_id",
 85            "rel_obj_id__isnull",
 86            "paused",
 87        )
 88
 89
 90class ScheduleViewSet(
 91    RetrieveModelMixin,
 92    UpdateModelMixin,
 93    ListModelMixin,
 94    GenericViewSet,
 95):
 96    queryset = (
 97        Schedule.objects.select_related("rel_obj_content_type")
 98        .defer("args", "kwargs", "options")
 99        .all()
100    )
101    serializer_class = ScheduleSerializer
102    search_fields = (
103        "id",
104        "identifier",
105        "_uid",
106        "actor_name",
107        "rel_obj_content_type__app_label",
108        "rel_obj_content_type__model",
109        "rel_obj_id",
110    )
111    filterset_class = ScheduleFilter
112    ordering = (
113        "next_run",
114        "actor_name",
115        "identifier",
116    )
117
118    @permission_required("authentik_tasks_schedules.send_schedule")
119    @extend_schema(
120        request=OpenApiTypes.NONE,
121        responses={
122            204: OpenApiResponse(description="Schedule sent successfully"),
123            404: OpenApiResponse(description="Schedule not found"),
124            500: OpenApiResponse(description="Failed to send schedule"),
125        },
126    )
127    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
128    def send(self, request: Request, pk=None) -> Response:
129        """Trigger this schedule now"""
130        schedule: Schedule = self.get_object()
131        schedule.send()
132        return Response({})
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
class ScheduleSerializer(authentik.core.api.utils.ModelSerializer): 
30class ScheduleSerializer(ModelSerializer):
31    rel_obj_app_label = ReadOnlyField(source="rel_obj_content_type.app_label")
32    rel_obj_model = ReadOnlyField(source="rel_obj_content_type.model")
33
34    description = SerializerMethodField()
35    last_task_status = SerializerMethodField()
36
37    class Meta:
38        model = Schedule
39        fields = (
40            "id",
41            "identifier",
42            "uid",
43            "actor_name",
44            "rel_obj_app_label",
45            "rel_obj_model",
46            "rel_obj_id",
47            "crontab",
48            "paused",
49            "next_run",
50            "description",
51            "last_task_status",
52        )
53
54    def get_description(self, instance: Schedule) -> str | None:
55        try:
56            actor: Actor = get_broker().get_actor(instance.actor_name)
57        except ActorNotFound:
58            LOGGER.warning("Could not find actor for schedule", schedule=instance)
59            return None
60        if "description" not in actor.options:
61            LOGGER.warning(
62                "Could not find description for actor",
63                schedule=instance,
64                actor=actor.actor_name,
65            )
66            return None
67        return actor.options["description"]
68
69    def get_last_task_status(self, instance: Schedule) -> TaskStatus | None:
70        last_task: Task = instance.tasks.defer("message", "result").order_by("-mtime").first()
71        if last_task:
72            return last_task.aggregated_status
73        return None

A ModelSerializer is just a regular Serializer, except that:

  • A set of default fields are automatically populated.
  • A set of default validators are automatically populated.
  • Default .create() and .update() implementations are provided.

The process of automatically determining a set of serializer fields based on the model fields is reasonably complex, but you almost certainly don't need to dig into the implementation.

If the ModelSerializer class doesn't generate the set of fields that you need you should either declare the extra/differing fields explicitly on the serializer class, or simply use a Serializer class.

rel_obj_app_label
rel_obj_model
description
last_task_status
def get_description(self, instance: authentik.tasks.schedules.models.Schedule) -> str | None: 
54    def get_description(self, instance: Schedule) -> str | None:
55        try:
56            actor: Actor = get_broker().get_actor(instance.actor_name)
57        except ActorNotFound:
58            LOGGER.warning("Could not find actor for schedule", schedule=instance)
59            return None
60        if "description" not in actor.options:
61            LOGGER.warning(
62                "Could not find description for actor",
63                schedule=instance,
64                actor=actor.actor_name,
65            )
66            return None
67        return actor.options["description"]
def get_last_task_status( self, instance: authentik.tasks.schedules.models.Schedule) -> authentik.tasks.models.TaskStatus | None: 
69    def get_last_task_status(self, instance: Schedule) -> TaskStatus | None:
70        last_task: Task = instance.tasks.defer("message", "result").order_by("-mtime").first()
71        if last_task:
72            return last_task.aggregated_status
73        return None
Inherited Members
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
class ScheduleSerializer.Meta: 
37    class Meta:
38        model = Schedule
39        fields = (
40            "id",
41            "identifier",
42            "uid",
43            "actor_name",
44            "rel_obj_app_label",
45            "rel_obj_model",
46            "rel_obj_id",
47            "crontab",
48            "paused",
49            "next_run",
50            "description",
51            "last_task_status",
52        )
fields = ('id', 'identifier', 'uid', 'actor_name', 'rel_obj_app_label', 'rel_obj_model', 'rel_obj_id', 'crontab', 'paused', 'next_run', 'description', 'last_task_status')
class ScheduleFilter(django_filters.filterset.FilterSet): 
76class ScheduleFilter(FilterSet):
77    rel_obj_id__isnull = BooleanFilter("rel_obj_id", "isnull")
78
79    class Meta:
80        model = Schedule
81        fields = (
82            "actor_name",
83            "rel_obj_content_type__app_label",
84            "rel_obj_content_type__model",
85            "rel_obj_id",
86            "rel_obj_id__isnull",
87            "paused",
88        )
rel_obj_id__isnull
declared_filters = OrderedDict({'rel_obj_id__isnull': <django_filters.filters.BooleanFilter object>})
base_filters = OrderedDict({'actor_name': <django_filters.filters.CharFilter object>, 'rel_obj_content_type__app_label': <django_filters.filters.CharFilter object>, 'rel_obj_content_type__model': <django_filters.filters.CharFilter object>, 'rel_obj_id': <django_filters.filters.CharFilter object>, 'rel_obj_id__isnull': <django_filters.filters.BooleanFilter object>, 'paused': <django_filters.filters.BooleanFilter object>})
class ScheduleFilter.Meta: 
79    class Meta:
80        model = Schedule
81        fields = (
82            "actor_name",
83            "rel_obj_content_type__app_label",
84            "rel_obj_content_type__model",
85            "rel_obj_id",
86            "rel_obj_id__isnull",
87            "paused",
88        )
fields = ('actor_name', 'rel_obj_content_type__app_label', 'rel_obj_content_type__model', 'rel_obj_id', 'rel_obj_id__isnull', 'paused')
class ScheduleViewSet(rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.UpdateModelMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): 
 91class ScheduleViewSet(
 92    RetrieveModelMixin,
 93    UpdateModelMixin,
 94    ListModelMixin,
 95    GenericViewSet,
 96):
 97    queryset = (
 98        Schedule.objects.select_related("rel_obj_content_type")
 99        .defer("args", "kwargs", "options")
100        .all()
101    )
102    serializer_class = ScheduleSerializer
103    search_fields = (
104        "id",
105        "identifier",
106        "_uid",
107        "actor_name",
108        "rel_obj_content_type__app_label",
109        "rel_obj_content_type__model",
110        "rel_obj_id",
111    )
112    filterset_class = ScheduleFilter
113    ordering = (
114        "next_run",
115        "actor_name",
116        "identifier",
117    )
118
119    @permission_required("authentik_tasks_schedules.send_schedule")
120    @extend_schema(
121        request=OpenApiTypes.NONE,
122        responses={
123            204: OpenApiResponse(description="Schedule sent successfully"),
124            404: OpenApiResponse(description="Schedule not found"),
125            500: OpenApiResponse(description="Failed to send schedule"),
126        },
127    )
128    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
129    def send(self, request: Request, pk=None) -> Response:
130        """Trigger this schedule now"""
131        schedule: Schedule = self.get_object()
132        schedule.send()
133        return Response({})

Retrieve a model instance.

queryset = <PostgresQuerySet []>
serializer_class = <class 'ScheduleSerializer'>
search_fields = ('id', 'identifier', '_uid', 'actor_name', 'rel_obj_content_type__app_label', 'rel_obj_content_type__model', 'rel_obj_id')
filterset_class = <class 'ScheduleFilter'>
ordering = ('next_run', 'actor_name', 'identifier')
@permission_required('authentik_tasks_schedules.send_schedule')
@extend_schema(request=OpenApiTypes.NONE, responses={204: OpenApiResponse(description='Schedule sent successfully'), 404: OpenApiResponse(description='Schedule not found'), 500: OpenApiResponse(description='Failed to send schedule')})
@action(detail=True, pagination_class=None, filter_backends=[], methods=['POST'])
def send( self, request: rest_framework.request.Request, pk=None) -> rest_framework.response.Response: 
119    @permission_required("authentik_tasks_schedules.send_schedule")
120    @extend_schema(
121        request=OpenApiTypes.NONE,
122        responses={
123            204: OpenApiResponse(description="Schedule sent successfully"),
124            404: OpenApiResponse(description="Schedule not found"),
125            500: OpenApiResponse(description="Failed to send schedule"),
126        },
127    )
128    @action(detail=True, pagination_class=None, filter_backends=[], methods=["POST"])
129    def send(self, request: Request, pk=None) -> Response:
130        """Trigger this schedule now"""
131        schedule: Schedule = self.get_object()
132        schedule.send()
133        return Response({})

Trigger this schedule now

name = None
description = None
suffix = None
detail = None
basename = None