authentik.tasks.schedules.models 

 1from django.apps import apps
 2from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 3from django.contrib.contenttypes.models import ContentType
 4from django.db import models
 5from django.utils.translation import gettext_lazy as _
 6from django_dramatiq_postgres.models import ScheduleBase
 7from psqlextra.manager import PostgresManager
 8
 9from authentik.lib.models import SerializerModel
10from authentik.tasks.models import TasksModel
11from authentik.tasks.schedules.common import ScheduleSpec
12
13
14class Schedule(TasksModel, SerializerModel, ScheduleBase):
15    identifier = models.TextField(
16        editable=False,
17        null=True,
18        help_text=_("Unique schedule identifier"),
19    )
20    _uid = models.TextField(
21        blank=True,
22        null=True,
23        help_text=_("User schedule identifier"),
24    )
25
26    rel_obj_content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True)
27    rel_obj_id = models.TextField(null=True)
28    rel_obj = GenericForeignKey("rel_obj_content_type", "rel_obj_id")
29
30    objects = PostgresManager()
31
32    class Meta(ScheduleBase.Meta):
33        default_permissions = (
34            "change",
35            "view",
36        )
37        permissions = [
38            ("send_schedule", _("Manually trigger a schedule")),
39        ]
40        indexes = (models.Index(fields=("rel_obj_content_type", "rel_obj_id")),)
41        unique_together = (
42            (
43                "actor_name",
44                "identifier",
45            ),
46        )
47
48    def __str__(self):
49        return f"Schedule {self.actor_name}:{self.uid}"
50
51    @property
52    def uid(self) -> str:
53        uid = str(self.actor_name)
54        if self._uid:
55            uid += f":{self._uid}"
56        return uid
57
58    @property
59    def serializer(self):
60        from authentik.tasks.schedules.api import ScheduleSerializer
61
62        return ScheduleSerializer
63
64
65class ScheduledModel(TasksModel, models.Model):
66    schedules = GenericRelation(
67        Schedule, content_type_field="rel_obj_content_type", object_id_field="rel_obj_id"
68    )
69
70    class Meta:
71        abstract = True
72
73    @classmethod
74    def models(cls) -> list[models.Model]:
75        def is_scheduled_model(klass) -> bool:
76            if ScheduledModel in klass.__bases__:
77                return True
78            return any(is_scheduled_model(klass) for klass in klass.__bases__)
79
80        return [
81            model
82            for model in apps.get_models()
83            if is_scheduled_model(model) and not model.__subclasses__()
84        ]
85
86    @property
87    def schedule_specs(self) -> list[ScheduleSpec]:
88        raise NotImplementedError
class Schedule(authentik.tasks.models.TasksModel, authentik.lib.models.SerializerModel, django_dramatiq_postgres.models.ScheduleBase): 
15class Schedule(TasksModel, SerializerModel, ScheduleBase):
16    identifier = models.TextField(
17        editable=False,
18        null=True,
19        help_text=_("Unique schedule identifier"),
20    )
21    _uid = models.TextField(
22        blank=True,
23        null=True,
24        help_text=_("User schedule identifier"),
25    )
26
27    rel_obj_content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True)
28    rel_obj_id = models.TextField(null=True)
29    rel_obj = GenericForeignKey("rel_obj_content_type", "rel_obj_id")
30
31    objects = PostgresManager()
32
33    class Meta(ScheduleBase.Meta):
34        default_permissions = (
35            "change",
36            "view",
37        )
38        permissions = [
39            ("send_schedule", _("Manually trigger a schedule")),
40        ]
41        indexes = (models.Index(fields=("rel_obj_content_type", "rel_obj_id")),)
42        unique_together = (
43            (
44                "actor_name",
45                "identifier",
46            ),
47        )
48
49    def __str__(self):
50        return f"Schedule {self.actor_name}:{self.uid}"
51
52    @property
53    def uid(self) -> str:
54        uid = str(self.actor_name)
55        if self._uid:
56            uid += f":{self._uid}"
57        return uid
58
59    @property
60    def serializer(self):
61        from authentik.tasks.schedules.api import ScheduleSerializer
62
63        return ScheduleSerializer

Schedule(id, actor_name, args, kwargs, options, crontab, paused, next_run, identifier, _uid, rel_obj_content_type, rel_obj_id)

def identifier(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

rel_obj_content_type

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def rel_obj_id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

rel_obj

Provide a generic many-to-one relation through the content_type and object_id fields.

This class also doubles as an accessor to the related object (similar to ForwardManyToOneDescriptor) by adding itself as a model attribute.

def objects(unknown):

The type of the None singleton.

uid: str 
52    @property
53    def uid(self) -> str:
54        uid = str(self.actor_name)
55        if self._uid:
56            uid += f":{self._uid}"
57        return uid
serializer 
59    @property
60    def serializer(self):
61        from authentik.tasks.schedules.api import ScheduleSerializer
62
63        return ScheduleSerializer

Get serializer for this model

tasks

Accessor to the related objects manager on the one-to-many relation created by GenericRelation.

In the example::

class Post(Model):
    comments = GenericRelation(Comment)

post.comments is a ReverseGenericManyToOneDescriptor instance.

def id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def actor_name(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def args(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def kwargs(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def options(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def crontab(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def paused(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def next_run(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

rel_obj_content_type_id
def get_next_by_next_run(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

def get_previous_by_next_run(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

Inherited Members
authentik.tasks.models.TasksModel
Meta
class Schedule.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class Schedule.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

class ScheduledModel(authentik.tasks.models.TasksModel, django.db.models.base.Model): 
66class ScheduledModel(TasksModel, models.Model):
67    schedules = GenericRelation(
68        Schedule, content_type_field="rel_obj_content_type", object_id_field="rel_obj_id"
69    )
70
71    class Meta:
72        abstract = True
73
74    @classmethod
75    def models(cls) -> list[models.Model]:
76        def is_scheduled_model(klass) -> bool:
77            if ScheduledModel in klass.__bases__:
78                return True
79            return any(is_scheduled_model(klass) for klass in klass.__bases__)
80
81        return [
82            model
83            for model in apps.get_models()
84            if is_scheduled_model(model) and not model.__subclasses__()
85        ]
86
87    @property
88    def schedule_specs(self) -> list[ScheduleSpec]:
89        raise NotImplementedError

Make subclasses preserve the alters_data attribute on overridden methods.

schedules

Accessor to the related objects manager on the one-to-many relation created by GenericRelation.

In the example::

class Post(Model):
    comments = GenericRelation(Comment)

post.comments is a ReverseGenericManyToOneDescriptor instance.

@classmethod
def models(unknown): 
74    @classmethod
75    def models(cls) -> list[models.Model]:
76        def is_scheduled_model(klass) -> bool:
77            if ScheduledModel in klass.__bases__:
78                return True
79            return any(is_scheduled_model(klass) for klass in klass.__bases__)
80
81        return [
82            model
83            for model in apps.get_models()
84            if is_scheduled_model(model) and not model.__subclasses__()
85        ]
schedule_specs: list[authentik.tasks.schedules.common.ScheduleSpec] 
87    @property
88    def schedule_specs(self) -> list[ScheduleSpec]:
89        raise NotImplementedError
tasks

Accessor to the related objects manager on the one-to-many relation created by GenericRelation.

In the example::

class Post(Model):
    comments = GenericRelation(Comment)

post.comments is a ReverseGenericManyToOneDescriptor instance.

class ScheduledModel.Meta: 
71    class Meta:
72        abstract = True
abstract = False