wip

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

authentik/admin/apps.py

@@ -39,7 +39,7 @@ class AuthentikAdminConfig(ManagedAppConfig):
 
         return [
             ScheduleSpec(
-                actor_name=update_latest_version.actor_name,
+                actor=update_latest_version,
                 crontab=f"{fqdn_rand('admin_latest_version')} * * * *",
             ),
         ]

authentik/blueprints/apps.py

@@ -150,14 +150,16 @@ class AuthentikBlueprintsConfig(ManagedAppConfig):
 
     @property
     def tenant_schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.blueprints.v1.tasks import blueprints_discovery, clear_failed_blueprints
+
         return [
             ScheduleSpec(
-                actor_name="authentik.blueprints.v1.tasks.blueprints_discovery",
+                actor=blueprints_discovery,
                 crontab=f"{fqdn_rand('blueprints_v1_discover')} * * * *",
                 run_on_startup=True,
             ),
             ScheduleSpec(
-                actor_name="authentik.blueprints.v1.tasks.clear_failed_blueprints",
+                actor=clear_failed_blueprints,
                 crontab=f"{fqdn_rand('blueprints_v1_cleanup')} * * * *",
                 run_on_startup=True,
             ),

authentik/blueprints/v1/tasks.py

@@ -90,13 +90,7 @@ class BlueprintEventHandler(FileSystemEventHandler):
         LOGGER.debug("new blueprint file created, starting discovery")
         for tenant in Tenant.objects.filter(ready=True):
             with tenant:
-                schedule = Schedule.objects.filter(
-                    actor_name=blueprints_discovery.actor_name,
-                    paused=False,
-                ).first()
-                if schedule:
-                    schedule.send()
-                # Schedule was paused or doesn't exist, no dispatch
+                Schedule.dispatch_by_actor(blueprints_discovery)
 
     def on_modified(self, event: FileSystemEvent):
         """Process file modification"""

authentik/core/apps.py

@@ -28,13 +28,15 @@ class AuthentikCoreConfig(ManagedAppConfig):
 
     @property
     def tenant_schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.core.tasks import clean_expired_models, clean_temporary_users
+
         return [
             ScheduleSpec(
-                actor_name="authentik.core.tasks.clean_expired_models",
+                actor=clean_expired_models,
                 crontab="2-59/5 * * * *",
             ),
             ScheduleSpec(
-                actor_name="authentik.core.tasks.clean_temporary_users",
+                actor=clean_temporary_users,
                 crontab="9-59/5 * * * *",
             ),
         ]

authentik/crypto/apps.py

@@ -72,9 +72,11 @@ class AuthentikCryptoConfig(ManagedAppConfig):
 
     @property
     def tenant_schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.crypto.tasks import certificate_discovery
+
         return [
             ScheduleSpec(
-                actor_name="authentik.crypto.tasks.certificate_discovery",
+                actor=certificate_discovery,
                 crontab=f"{fqdn_rand('crypto_certificate_discovery')} * * * *",
             ),
         ]

authentik/enterprise/apps.py

@@ -35,7 +35,7 @@ class AuthentikEnterpriseConfig(EnterpriseConfig):
 
         return [
             ScheduleSpec(
-                actor_name=enterprise_update_usage.actor_name,
+                actor=enterprise_update_usage,
                 crontab=f"{fqdn_rand('enterprise_update_usage')} */2 * * *",
             ),
         ]

authentik/enterprise/policies/unique_password/apps.py

@@ -20,11 +20,11 @@ class AuthentikEnterprisePoliciesUniquePasswordConfig(EnterpriseConfig):
 
         return [
             ScheduleSpec(
-                actor_name=trim_password_histories.actor_name,
+                actor=trim_password_histories,
                 crontab=f"{fqdn_rand('policies_unique_password_trim')} */12 * * *",
             ),
             ScheduleSpec(
-                actor_name=check_and_purge_password_history.actor_name,
+                actor=check_and_purge_password_history,
                 crontab=f"{fqdn_rand('policies_unique_password_purge')} */24 * * *",
             ),
         ]

authentik/enterprise/providers/google_workspace/models.py

@@ -7,6 +7,7 @@ from django.db import models
 from django.db.models import QuerySet
 from django.templatetags.static import static
 from django.utils.translation import gettext_lazy as _
+from dramatiq.actor import Actor
 from google.oauth2.service_account import Credentials
 from rest_framework.serializers import Serializer
 
@@ -112,8 +113,10 @@ class GoogleWorkspaceProvider(OutgoingSyncProvider, ScheduledModel, BackchannelP
     )
 
     @property
-    def sync_task(self) -> str:
-        return "authentik.enterprise.providers.google_workspace.tasks.google_workspace_sync"
+    def sync_actor(self) -> Actor:
+        from authentik.enterprise.providers.google_workspace.tasks import google_workspace_sync
+
+        return google_workspace_sync
 
     def client_for_model(
         self,

authentik/enterprise/providers/microsoft_entra/models.py

@@ -8,6 +8,7 @@ from django.db import models
 from django.db.models import QuerySet
 from django.templatetags.static import static
 from django.utils.translation import gettext_lazy as _
+from dramatiq.actor import Actor
 from rest_framework.serializers import Serializer
 
 from authentik.core.models import (
@@ -101,8 +102,10 @@ class MicrosoftEntraProvider(OutgoingSyncProvider, ScheduledModel, BackchannelPr
     )
 
     @property
-    def sync_task(self) -> str:
-        return "authentik.enterprise.providers.microsoft_entra.tasks.microsoft_entra_sync"
+    def sync_actor(self) -> Actor:
+        from authentik.enterprise.providers.microsoft_entra.tasks import microsoft_entra_sync
+
+        return microsoft_entra_sync
 
     def client_for_model(
         self,

authentik/lib/sync/outgoing/models.py

@@ -1,5 +1,6 @@
 from typing import Any, Self
 
+from dramatiq.actor import Actor
 import pglock
 from django.core.paginator import Paginator
 from django.db import connection, models
@@ -60,14 +61,14 @@ class OutgoingSyncProvider(ScheduledModel, Model):
         )
 
     @property
-    def sync_task(self) -> str:
+    def sync_actor(self) -> Actor:
         raise NotImplementedError
 
     @property
     def schedule_specs(self) -> list[ScheduleSpec]:
         return [
             ScheduleSpec(
-                actor_name=self.sync_task,
+                actor=self.sync_actor,
                 uid=self.pk,
                 args=(self.pk,),
                 options={

authentik/outposts/apps.py

@@ -69,7 +69,7 @@ class AuthentikOutpostConfig(ManagedAppConfig):
 
         return [
             ScheduleSpec(
-                actor_name=outpost_token_ensurer.actor_name,
+                actor=outpost_token_ensurer,
                 crontab=f"{fqdn_rand('outpost_token_ensurer')} */8 * * *",
             ),
         ]
@@ -80,7 +80,7 @@ class AuthentikOutpostConfig(ManagedAppConfig):
 
         return [
             ScheduleSpec(
-                actor_name=outpost_connection_discovery.actor_name,
+                actor=outpost_connection_discovery,
                 crontab=f"{fqdn_rand('outpost_connection_discovery')} */8 * * *",
                 run_on_startup=True,
             ),

authentik/outposts/models.py

@@ -169,7 +169,7 @@ class OutpostServiceConnection(ScheduledModel, models.Model):
 
         return [
             ScheduleSpec(
-                actor_name=outpost_service_connection_monitor.actor_name,
+                actor=outpost_service_connection_monitor,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab="3-59/15 * * * *",
@@ -321,7 +321,7 @@ class Outpost(ScheduledModel, SerializerModel, ManagedModel):
 
         return [
             ScheduleSpec(
-                actor_name=outpost_controller.actor_name,
+                actor=outpost_controller,
                 uid=self.pk,
                 args=(self.pk,),
                 kwargs={"action": "up", "from_cache": False},

authentik/providers/scim/models.py

@@ -7,6 +7,7 @@ from django.db import models
 from django.db.models import QuerySet
 from django.templatetags.static import static
 from django.utils.translation import gettext_lazy as _
+from dramatiq.actor import Actor
 from rest_framework.serializers import Serializer
 
 from authentik.core.models import BackchannelProvider, Group, PropertyMapping, User, UserTypes
@@ -101,8 +102,10 @@ class SCIMProvider(OutgoingSyncProvider, ScheduledModel, BackchannelProvider):
         return static("authentik/sources/scim.png")
 
     @property
-    def sync_task(self) -> str:
-        return "authentik.providers.scim.tasks.scim_sync"
+    def sync_actor(self) -> Actor:
+        from authentik.providers.scim.tasks import scim_sync
+
+        return scim_sync
 
     def client_for_model(
         self, model: type[User | Group | SCIMProviderUser | SCIMProviderGroup]

authentik/sources/kerberos/models.py

@@ -140,16 +140,18 @@ class KerberosSource(ScheduledModel, Source):
 
     @property
     def schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.sources.kerberos.tasks import kerberos_sync, kerberos_connectivity_check
+
         return [
             ScheduleSpec(
-                actor_name="authentik.sources.kerberos.tasks.kerberos_sync",
+                actor=kerberos_sync,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab=f"{fqdn_rand('kerberos_sync/' + str(self.pk))} */2 * * *",
                 description=_(f"Sync Kerberos source '{self.name}'"),
             ),
             ScheduleSpec(
-                actor_name="authentik.sources.kerberos.tasks.kerberos_connectivity_check",
+                actor=kerberos_connectivity_check,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab=f"{fqdn_rand('kerberos_connectivity_check/' + str(self.pk))} * * * *",

authentik/sources/ldap/models.py

@@ -164,16 +164,18 @@ class LDAPSource(ScheduledModel, Source):
 
     @property
     def schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.sources.ldap.tasks import ldap_sync, ldap_connectivity_check
+
         return [
             ScheduleSpec(
-                actor_name="authentik.sources.ldap.tasks.ldap_sync",
+                actor=ldap_sync,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab=f"{fqdn_rand('ldap_sync/' + str(self.pk))} */2 * * *",
                 description=_(f"Sync LDAP source '{self.name}'"),
             ),
             ScheduleSpec(
-                actor_name="authentik.sources.ldap.tasks.ldap_connectivity_check",
+                actor=ldap_connectivity_check,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab=f"{fqdn_rand('ldap_connectivity_check/' + str(self.pk))} * * * *",

authentik/sources/oauth/apps.py

@@ -45,9 +45,11 @@ class AuthentikSourceOAuthConfig(ManagedAppConfig):
 
     @property
     def tenant_schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.sources.oauth.tasks import update_well_known_jwks
+
         return [
             ScheduleSpec(
-                actor_name="authentik.sources.oauth.tasks.update_well_known_jwks",
+                actor=update_well_known_jwks,
                 crontab=f"{fqdn_rand('update_well_known_jwks')} */3 * * *",
             ),
         ]

authentik/sources/plex/models.py

@@ -77,9 +77,11 @@ class PlexSource(ScheduledModel, Source):
 
     @property
     def schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.sources.plex.tasks import check_plex_token
+
         return [
             ScheduleSpec(
-                actor_name="authentik.sources.plex.tasks.check_plex_token",
+                actor=check_plex_token,
                 uid=self.pk,
                 args=(self.pk,),
                 crontab=f"{fqdn_rand(self.pk)} */3 * * *",

authentik/stages/authenticator_webauthn/apps.py

@@ -15,9 +15,11 @@ class AuthentikStageAuthenticatorWebAuthnConfig(ManagedAppConfig):
 
     @property
     def tenant_schedule_specs(self) -> list[ScheduleSpec]:
+        from authentik.stages.authenticator_webauthn.tasks import webauthn_mds_import
+
         return [
             ScheduleSpec(
-                actor_name="authentik.stages.authenticator_webauthn.tasks.webauthn_mds_import",
+                actor=webauthn_mds_import,
                 crontab=f"{fqdn_rand('webauthn_mds_import')} {fqdn_rand('webauthn_mds_import', 24)} * * {fqdn_rand('webauthn_mds_import', 7)}",  # noqa: E501
             ),
         ]

authentik/tasks/schedules/lib.py

@@ -3,13 +3,15 @@ from collections.abc import Iterable
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 
+from dramatiq.actor import Actor
+
 if TYPE_CHECKING:
     from authentik.tasks.schedules.models import Schedule
 
 
 @dataclass
 class ScheduleSpec:
-    actor_name: str
+    actor: Actor
     crontab: str
     uid: str | None = None
 
@@ -27,8 +29,8 @@ class ScheduleSpec:
 
     def get_uid(self) -> str:
         if self.uid is not None:
-            return f"{self.actor_name}:{self.uid}"
-        return self.actor_name
+            return f"{self.actor.actor_name}:{self.uid}"
+        return self.actor.actor_name
 
     def get_args(self) -> bytes:
         return pickle.dumps(self.args)
@@ -47,7 +49,7 @@ class ScheduleSpec:
         }
         defaults = {
             **query,
-            "actor_name": self.actor_name,
+            "actor_name": self.actor.actor_name,
             "args": self.get_args(),
             "kwargs": self.get_kwargs(),
             "options": self.get_options(),