Expose TaskSchedule in plugin API and add args/kwargs

CHANGES/plugin_api/7608.feature

@@ -0,0 +1,4 @@
+Added `task_args` and `task_kwargs` fields to `TaskSchedule`, allowing plugins to store positional
+and keyword arguments that are forwarded to tasks when they are dispatched on schedule. Both fields
+are encrypted at rest. `TaskSchedule` and `TaskScheduleSerializer` are now exposed via the plugin
+API.

docs/dev/learn/other/task-scheduling.md

@@ -14,3 +14,26 @@ therefore scheduled task dispatching will be missed if all workers are offline.
 window, overdue schedules will dispatch at most one task, but down to timing, they may be
 rescheduled shortly thereafter. The task schedule API at `/pulp/api/v3/task-schedules/` is
 provided to read the tasks schedules.
+
+## Passing Arguments to Scheduled Tasks
+
+The `task_args` and `task_kwargs` fields on `TaskSchedule` allow plugin writers to store positional
+and keyword arguments that will be forwarded to the task function each time it is dispatched. This
+is useful when a scheduled task needs to operate on a specific resource or with specific options.
+
+```python
+from datetime import timedelta
+from pulpcore.plugin.models import TaskSchedule
+
+TaskSchedule(
+    name="my-plugin-sync-schedule",
+    task_name="my_plugin.app.tasks.sync",
+    task_kwargs={"remote_pk": str(remote.pk), "optimize": True},
+    dispatch_interval=timedelta(hours=6),
+).save()
+```
+
+The args and kwargs stored in `task_args` and `task_kwargs` are passed directly
+to `dispatch()` when the schedule fires, so they should match the signature of
+the task function referenced by `task_name`. Both fields are encrypted at rest
+(they will not be included in API responses).

pulpcore/app/migrations/0150_taskschedule_task_kwargs.py

@@ -0,0 +1,24 @@
+# Generated manually
+
+from django.db import migrations
+
+import pulpcore.app.models.fields
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0149_distributedpublication"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="taskschedule",
+            name="task_args",
+            field=pulpcore.app.models.fields.EncryptedJSONField(default=list),
+        ),
+        migrations.AddField(
+            model_name="taskschedule",
+            name="task_kwargs",
+            field=pulpcore.app.models.fields.EncryptedJSONField(default=dict),
+        ),
+    ]

pulpcore/app/models/task.py

@@ -443,6 +443,8 @@ class TaskSchedule(BaseModel):
     next_dispatch = models.DateTimeField(default=timezone.now, null=True)
     dispatch_interval = models.DurationField(null=True)
     task_name = models.TextField()
+    task_args = EncryptedJSONField(default=list)
+    task_kwargs = EncryptedJSONField(default=dict)
     last_task = models.ForeignKey(Task, null=True, on_delete=models.SET_NULL)
 
     class Meta:

pulpcore/plugin/models/__init__.py

@@ -38,6 +38,7 @@
     SigningService,
     Task,
     TaskGroup,
+    TaskSchedule,
     Upload,
     UploadChunk,
     VulnerabilityReport,
@@ -84,6 +85,7 @@
     "SigningService",
     "Task",
     "TaskGroup",
+    "TaskSchedule",
     "Upload",
     "UploadChunk",
     "EncryptedTextField",

pulpcore/plugin/serializers/__init__.py

@@ -38,6 +38,7 @@
     SingleArtifactContentSerializer,
     SingleContentArtifactField,
     TaskGroupOperationResponseSerializer,
+    TaskScheduleSerializer,
     RepositoryAddRemoveContentSerializer,
     ValidateFieldsMixin,
     validate_unknown_fields,
@@ -89,6 +90,7 @@
     "SingleArtifactContentSerializer",
     "SingleContentArtifactField",
     "TaskGroupOperationResponseSerializer",
+    "TaskScheduleSerializer",
     "RepositoryAddRemoveContentSerializer",
     "ValidateFieldsMixin",
     "validate_unknown_fields",

pulpcore/tasking/_util.py

@@ -300,6 +300,8 @@ def dispatch_scheduled_tasks():
             with transaction.atomic():
                 task_schedule.last_task = dispatch(
                     task_schedule.task_name,
+                    args=task_schedule.task_args,
+                    kwargs=task_schedule.task_kwargs,
                 )
                 task_schedule.save(update_fields=["next_dispatch", "last_task"])