feat: Add command to schedule resource sync task

galaxy_ng/app/management/commands/task-scheduler.py

@@ -0,0 +1,61 @@
+import importlib
+from django.core.management.base import BaseCommand, CommandError
+from datetime import timedelta
+from django.utils.timezone import now
+
+
+class Command(BaseCommand):
+    """Schedules a task for execution using Pulp Tasking System."""
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--id',
+            required=True,
+            type=str,
+            help="Unique str identifier for scheduled task e.g: make_sandwich"
+        )
+        parser.add_argument(
+            '--path',
+            required=True,
+            help="Importable path for the callable e.g: galaxy_ng.app.foo.bar"
+        )
+        parser.add_argument(
+            '--interval',
+            required=True,
+            type=int,
+            help="Interval in minutes"
+        )
+        parser.add_argument(
+            '--force',
+            action="store_true",
+            default=False,
+            help="Override existing scheduled task with the same identifier"
+        )
+
+    def handle(self, *args, **options):
+        # bypass pulp bad import check because the model is not exposed on plugins path
+        TaskSchedule = importlib.import_module("pulpcore.app.models").TaskSchedule
+        identifier = options["id"]
+        function_path = options["path"]
+        dispatch_interval = timedelta(minutes=options["interval"])
+        next_dispatch = now() + dispatch_interval
+
+        if existing := TaskSchedule.objects.filter(name=identifier):
+            if options["force"]:
+                existing.delete()
+            else:
+                raise CommandError(
+                    f"{identifier} is already scheduled, use --force to override it."
+                )
+
+        task = TaskSchedule(
+            name=identifier,
+            task_name=function_path,
+            dispatch_interval=dispatch_interval,
+            next_dispatch=next_dispatch
+        )
+        task.save()
+        self.stdout.write(
+            f"{task.name} scheduled for every {dispatch_interval} minutes. "
+            f"next execution on: {next_dispatch}"
+        )

galaxy_ng/app/tasks/resource_sync.py

@@ -0,0 +1,9 @@
+from pprint import pprint
+
+
+def run():  # pragma: no cover
+    """Start DAB Resource Sync"""
+    from ansible_base.resource_registry.tasks.sync import SyncExecutor
+    executor = SyncExecutor(retries=3)
+    executor.run()
+    pprint(executor.results)

galaxy_ng/tests/unit/app/management/commands/test_task_scheduler.py

@@ -0,0 +1,46 @@
+import importlib
+from io import StringIO
+from django.core.management import call_command, CommandError
+from django.test import TestCase
+from datetime import timedelta
+
+
+def make_sandwich():
+    """make a sandwich task"""
+    return "<bread>(picles)(lettuce)(onion)(tomato)(tofu)</bread>"
+
+
+FUNC_NAME = make_sandwich.__name__
+FUNC_PATH = f"{make_sandwich.__module__}.{FUNC_NAME}"
+
+
+class TestTaskScheduler(TestCase):
+    def setUp(self):
+        super().setUp()
+
+    def test_command_output(self):
+        with self.assertRaisesMessage(
+            CommandError, 'Error: the following arguments are required: --id, --path, --interval'
+        ):
+            call_command('task-scheduler')
+
+    def test_schedule_a_task(self):
+        out = StringIO()
+        call_command(
+            'task-scheduler',
+            '--id',
+            FUNC_NAME,
+            '--path',
+            FUNC_PATH,
+            '--interval',
+            '45',
+            stdout=out
+        )
+        self.assertIn(
+            f"{FUNC_NAME} scheduled for every 0:45:00 minutes.",
+            out.getvalue()
+        )
+        TaskSchedule = importlib.import_module("pulpcore.app.models").TaskSchedule
+        task = TaskSchedule.objects.get(name=FUNC_NAME)
+        assert task.dispatch_interval == timedelta(minutes=45)
+        assert task.task_name == FUNC_PATH