Pass through args and kwargs

Author: AlanCoding

ansible_base/task/__init__.py

@@ -13,7 +13,10 @@ def get_config():
                 "pg_notify": psycopg_params,
                 "channels": ["dab_broadcast"],
             },
-            "scheduled": {},
+            "scheduled": {
+                "ansible_base.task.tasks.run_task_from_queue": {"schedule": 60},
+                "ansible_base.task.tasks.manage_lost_tasks": {"schedule": 60*10}
+            },
         },
         "pool": {"max_workers": 4},  # TODO: to settings
     }

ansible_base/task/migrations/0003_task_args_task_kwargs.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.2.16 on 2025-02-18 18:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dab_tas', '0002_task_wrapper_uuid_alter_task_uuid'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='task',
+            name='args',
+            field=models.JSONField(blank=True, default=dict, help_text='The arguments for the task run.'),
+        ),
+        migrations.AddField(
+            model_name='task',
+            name='kwargs',
+            field=models.JSONField(blank=True, default=dict, help_text='The keyword arguments for the task run.'),
+        ),
+    ]

ansible_base/task/models.py

@@ -34,6 +34,13 @@ class Task(models.Model):
     )
     name = models.TextField(help_text=_('Importable path for class or method'))
 
+    args = models.JSONField(
+        default=dict, null=False, blank=True, help_text=_("The arguments for the task run.")
+    )
+    kwargs = models.JSONField(
+        default=dict, null=False, blank=True, help_text=_("The keyword arguments for the task run.")
+    )
+
     created = models.DateTimeField(
         auto_now_add=True, help_text=_('Time the publisher (submitter) of this task call created it, approximately the time of submission as well')
     )

ansible_base/task/publish.py

@@ -17,7 +17,7 @@ def task_name(self):
 
     def apply_async(self, args=None, kwargs=None):
         # this function may allow additional arguments in the future, but not now
-        Task.objects.create(name=self.task_name)
+        Task.objects.create(name=self.task_name, args=args, kwargs=kwargs)
 
         # pg_notify message (probably) to wake up
         transaction.on_commit(run_task_from_queue.delay)

ansible_base/task/tasks.py

@@ -2,9 +2,10 @@
 
 from dispatcher.utils import resolve_callable
 from dispatcher.publish import task
+from dispatcher.control import Control
 
 from django.db import transaction
-from django.utils.timezone import now
+from django.utils.timezone import now, timedelta
 
 from ansible_base.task.models import TASK_STATES, Task
 
@@ -29,13 +30,22 @@ def run_task_from_queue(dispatcher):
 
     try:
         task_callable = resolve_callable(task.name)
-        task_callable()
+        task_callable(*task.args, **task.kwargs)
     except Exception:
         logger.traceback(f'Failed to run and complete {task.name}')
 
     task.delete()
 
 
+@task(queue='dab_broadcast')
+def manage_lost_tasks(grace_period: int = 10):
+    cutoff_time = now() - timedelta(minutes=grace_period)
+    ctl = Control('dab_broadcast')
+    for task in Task.objects.filter(state=TASK_STATES.RUNNING, started_at__lt=cutoff_time).iterator():
+        running_tasks = ctl.control_with_reply('running', data={'uuid': task.wrapper_uuid})
+        print(running_tasks)
+        print(task.wrapper_uuid)
+
 # TODO: Add a "reaper" fallback task, this is currently blocked on dispatcher issue
 # https://github.com/ansible/dispatcher/issues/6
 # When run_task_from_queue starts the task, the uuid needs to be setup to be discoverable