Init migration, connect things

Author: AlanCoding

ansible_base/task/management/commands/run_dispatcher.py

@@ -1,23 +1,43 @@
+import asyncio
+import logging
+
 from dispatcher.main import DispatcherMain
+
+from ansible_base.lib.utils.db import get_pg_notify_params
+
 from django.core.management.base import BaseCommand
+from django.db import connection
+from django.core.cache import cache
+
+logger = logging.getLogger(__name__)
 
 
 class Command(BaseCommand):
     help = "Runs bug checking sanity checks, gets scale metrics, and recommendations for Role Based Access Control"
 
     def handle(self, *args, **options):
+        psycopg_params = get_pg_notify_params()
+
         dispatcher_config = {
             "producers": {
                 "brokers": {
-                    "pg_notify": {"conninfo": settings.PG_NOTIFY_DSN_SERVER},
-                    # TODO: sanitize or escape channel names on dispatcher side
+                    "pg_notify": psycopg_params,
                     "channels": ["dab_broadcast"],
                 },
-                # NOTE: I would prefer to move the activation monitoring
-                # from worker to activation, but that is more work
                 "scheduled": {},
             },
-            "pool": {"max_workers": 4},
+            "pool": {"max_workers": 4},  # TODO: to settings
         }
 
-        DispatcherMain()
+        loop = asyncio.get_event_loop()
+        dispatcher = DispatcherMain(dispatcher_config)
+
+        # Borrowed from eda-server, ensure the database connection is closed
+        connection.close()
+        cache.close()
+        try:
+            loop.run_until_complete(dispatcher.main())
+        except KeyboardInterrupt:
+            logger.info("run_worker_dispatch entry point leaving")
+        finally:
+            loop.close()

ansible_base/task/migrations/0001_initial.py

@@ -0,0 +1,26 @@
+# Generated by Django 4.2.16 on 2025-02-04 21:22
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Task',
+            fields=[
+                ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, help_text='UUID that corresponds to the dispatcher task uuid', primary_key=True, serialize=False)),
+                ('state', models.CharField(choices=[('completed', 'Completed'), ('failed', 'Failed'), ('running', 'Running'), ('waiting', 'Waiting')], default='waiting', help_text='Choices of this field track with acknowledgement and completion of a task', max_length=15)),
+                ('name', models.TextField(help_text='Importable path for class or method')),
+                ('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')),
+                ('started_at', models.DateTimeField(help_text='Time of acknowledgement, also approximately the time the task starts', null=True)),
+                ('finished_at', models.DateTimeField(help_text='Time task is cleared (whether failed or succeeded), may be unused if set to auto-delete', null=True)),
+            ],
+        ),
+    ]

ansible_base/task/models.py

@@ -28,6 +28,7 @@ class Task(models.Model):
     state = models.CharField(
         choices=[(s, s.title()) for s in sorted(vars(TASK_STATES).values())],
         default=TASK_STATES.WAITING,
+        max_length=15,
         help_text=_('Choices of this field track with acknowledgement and completion of a task'),
     )
     name = models.TextField(help_text=_('Importable path for class or method'))

test_app/settings.py

@@ -69,6 +69,7 @@
     'ansible_base.rest_pagination',
     'ansible_base.rbac',
     'ansible_base.oauth2_provider',
+    'ansible_base.task',
     'test_app',
     'django_extensions',
     'debug_toolbar',