Fix linting issues

Author: AlanCoding

ansible_base/task/admin.py

@@ -1,3 +0,0 @@
-from django.contrib import admin
-
-# Register your models here.

ansible_base/task/apps.py

@@ -1,6 +1,5 @@
-from django.apps import AppConfig
-
 from dispatcher.config import setup
+from django.apps import AppConfig
 
 from ansible_base.task.config import get_config
 

ansible_base/task/config.py

@@ -16,14 +16,14 @@ def get_config() -> dict:
                 "channels": settings.DAB_TASK_LISTEN_QUEUES,
                 # The default publish channel allows using .delay without more arguments
                 # this may still have task-specific overrides
-                "default_publish_channel": settings.DAB_TASK_ADMIN_QUEUE
+                "default_publish_channel": settings.DAB_TASK_ADMIN_QUEUE,
             }
         },
         "producers": {
             "ScheduledProducer": {
                 "task_schedule": {
                     "ansible_base.task.tasks.run_task_from_queue": {"schedule": 60},
-                    "ansible_base.task.tasks.manage_lost_tasks": {"schedule": 60*10}
+                    "ansible_base.task.tasks.manage_lost_tasks": {"schedule": 60 * 10},
                 }
             },
         },

ansible_base/task/management/commands/run_dispatcher.py

@@ -1,10 +1,9 @@
 import logging
 
 from dispatcher import run_service
-
+from django.core.cache import cache
 from django.core.management.base import BaseCommand
 from django.db import connection
-from django.core.cache import cache
 
 logger = logging.getLogger(__name__)
 

ansible_base/task/models.py

@@ -27,7 +27,9 @@ class Task(models.Model):
     uuid = models.UUIDField(primary_key=True, default=uuid4, editable=False, help_text=_('Original dispatcher UUID generated at time of task publishing'))
     # TODO: change back to UUIDField, workaround dispatcher using non-UUID uuids for now
     # null=True
-    wrapper_uuid = models.CharField(blank=True, default='', max_length=40, editable=False, help_text=_('Adopted UUID of DAB task app wrapper, attached when background service starts task'))
+    wrapper_uuid = models.CharField(
+        blank=True, default='', max_length=40, editable=False, help_text=_('Adopted UUID of DAB task app wrapper, attached when background service starts task')
+    )
     state = models.CharField(
         choices=[(s, s.title()) for s in sorted(vars(TASK_STATES).values())],
         default=TASK_STATES.WAITING,
@@ -37,12 +39,8 @@ class Task(models.Model):
     name = models.TextField(help_text=_('Importable path for class or method'))
     queue = models.TextField(default='', help_text=_('The work queue this task is submitted to, only nodes listening to this queue may process the task'))
 
-    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.")
-    )
+    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

@@ -1,6 +1,6 @@
 from dispatcher.utils import serialize_task
-from django.db import transaction
 from django.conf import settings
+from django.db import transaction
 
 from ansible_base.task.models import Task
 from ansible_base.task.tasks import run_task_from_queue

ansible_base/task/tasks.py

@@ -1,27 +1,26 @@
 import logging
 
-from dispatcher.utils import resolve_callable
-from dispatcher.publish import task
 from dispatcher.factories import get_control_from_settings
-
-from django.db import transaction
+from dispatcher.publish import task as dispatcher_task
+from dispatcher.utils import resolve_callable
 from django.conf import settings
+from django.db import transaction
 from django.utils.timezone import now, timedelta
 
 from ansible_base.task.models import TASK_STATES, Task
 
 logger = logging.getLogger(__name__)
 
 
-@task(queue=settings.DAB_TASK_ADMIN_QUEUE, bind=True)
+@dispatcher_task(queue=settings.DAB_TASK_ADMIN_QUEUE, bind=True)
 def run_task_from_queue(dispatcher):
     with transaction.atomic():
-        task = Task.objects.filter(state=TASK_STATES.WAITING, queue__in=settings.DAB_TASK_LISTEN_QUEUES).select_for_update().first()
-        if task:
-            task.state = TASK_STATES.RUNNING
-            task.started_at = now()
-            task.wrapper_uuid = str(dispatcher.uuid)
-            task.save(update_fields=['state', 'started_at', 'wrapper_uuid'])
+        db_task = Task.objects.filter(state=TASK_STATES.WAITING, queue__in=settings.DAB_TASK_LISTEN_QUEUES).select_for_update().first()
+        if db_task:
+            db_task.state = TASK_STATES.RUNNING
+            db_task.started_at = now()
+            db_task.wrapper_uuid = str(dispatcher.uuid)
+            db_task.save(update_fields=['state', 'started_at', 'wrapper_uuid'])
         else:
             return
 
@@ -30,26 +29,26 @@ def run_task_from_queue(dispatcher):
         run_task_from_queue.delay()
 
     try:
-        task_callable = resolve_callable(task.name)
-        task_callable(*task.args, **task.kwargs)
+        task_callable = resolve_callable(db_task.name)
+        task_callable(*db_task.args, **db_task.kwargs)
     except Exception:
-        logger.exception(f'Failed to run and complete {task.name}')
+        logger.exception(f'Failed to run and complete {db_task.name}')
 
-    task.delete()
+    db_task.delete()
 
 
-@task(queue=settings.DAB_TASK_ADMIN_QUEUE)
+@dispatcher_task(queue=settings.DAB_TASK_ADMIN_QUEUE)
 def manage_lost_tasks(grace_period: int = 10):
     cutoff_time = now() - timedelta(minutes=grace_period)
-    for task in Task.objects.filter(state=TASK_STATES.RUNNING, started_at__lt=cutoff_time, queue__in=settings.DAB_TASK_LISTEN_QUEUES).iterator():
+    for db_task in Task.objects.filter(state=TASK_STATES.RUNNING, started_at__lt=cutoff_time, queue__in=settings.DAB_TASK_LISTEN_QUEUES).iterator():
         ctl = get_control_from_settings(default_publish_channel=settings.DAB_TASK_ADMIN_QUEUE)
 
-        running_tasks = ctl.control_with_reply('running', data={'uuid': str(task.wrapper_uuid)})
+        running_tasks = ctl.control_with_reply('running', data={'uuid': str(db_task.wrapper_uuid)})
 
         found = False
         for server_reply in running_tasks:
             for worker_id, task_data in server_reply:
-                if task_data.get('uuid') == str(task.wrapper_uuid):
+                if task_data.get('uuid') == str(db_task.wrapper_uuid):
                     found = True
                     break
             if found:
@@ -59,11 +58,11 @@ def manage_lost_tasks(grace_period: int = 10):
             # TODO: feature of retry policy
             try:
                 with transaction.atomic():
-                    task = Task.objects.select_for_update().get(uuid=task.uuid)
-                    logger.warning(f'Could not find task {task.name} {task.wrapper_uuid}, deleting entry')
-                    task.delete()
+                    db_task = Task.objects.select_for_update().get(uuid=db_task.uuid)
+                    logger.warning(f'Could not find task {db_task.name} {db_task.wrapper_uuid}, deleting entry')
+                    db_task.delete()
             except Task.DoesNotExist:
-                logger.debug(f'task {task.name} uuid={task.uuid} already deleted, doing nothing')
+                logger.debug(f'task {db_task.name} uuid={db_task.uuid} already deleted, doing nothing')
         else:
-            delta = now() - task.started_at
-            logger.info(f'Noticed {task.name} {task.wrapper_uuid} running for {delta} seconds, seems to be fine')
+            delta = now() - db_task.started_at
+            logger.info(f'Noticed {db_task.name} {db_task.wrapper_uuid} running for {delta} seconds, seems to be fine')

test_app/tasks.py

@@ -1,9 +1,8 @@
 import time
 from uuid import UUID
 
-from test_app.models import UUIDModel
-
 from ansible_base.task.publish import durable_task
+from test_app.models import UUIDModel
 
 
 @durable_task()

test_app/tests/task/conftest.py

@@ -1,16 +1,13 @@
 import asyncio
-import multiprocessing
 import contextlib
+import multiprocessing
 import time
 from uuid import uuid4
 
 import pytest
-
-
 from dispatcher.main import DispatcherMain
 
 
-
 async def asyncio_target(queue_in, queue_out, config, this_uuid):
     try:
         dispatcher = DispatcherMain(config)
@@ -20,7 +17,6 @@ async def asyncio_target(queue_in, queue_out, config, this_uuid):
         await dispatcher.wait_for_producers_ready()
         queue_out.put('ready')
 
-
         print(f'{this_uuid} dispatcher server listening on queue_in')
         loop = asyncio.get_event_loop()
         message = await loop.run_in_executor(None, queue_in.get)

test_app/tests/task/test_basic.py

@@ -1,17 +1,16 @@
-from uuid import uuid4
 import time
+from uuid import uuid4
 
 import pytest
 
 from ansible_base.task.config import get_config
-
-from test_app.tasks import create_uuid_entry
 from test_app.models import UUIDModel
+from test_app.tasks import create_uuid_entry
 
 
 @pytest.mark.django_db
 def test_run_task(dispatcher_subprocess):
-    with dispatcher_subprocess(get_config()) as server:
+    with dispatcher_subprocess(get_config()):
         my_uuid = str(uuid4())
         create_uuid_entry.delay(uuid=my_uuid)