Demo via a test_app view

Author: AlanCoding

ansible_base/task/publish.py

@@ -37,3 +37,7 @@ def __call__(self, fn):
         setattr(fn, 'delay', publisher.delay)
 
         return fn
+
+
+def task():
+    return TaskDecorator()

requirements/requirements_dev.txt

@@ -21,4 +21,5 @@ pytest-django
 setuptools-scm
 sqlparse==0.5.2
 psycopg[binary]
+honcho
 sdb

test_app/urls.py

@@ -20,6 +20,7 @@
     path('api/v1/', include(resource_api_urls)),
     path('api/v1/', views.api_root),
     path('api/v1/timeout_view/', views.timeout_view, name='test-timeout-view'),
+    path('api/v1/trigger_task/', views.trigger_task, name='test-trigger-task'),
     path('login/', include('rest_framework.urls')),
     path("__debug__/", include("debug_toolbar.urls")),
 ]

test_app/views.py

@@ -5,10 +5,13 @@
 from django.shortcuts import render
 from django.urls.exceptions import NoReverseMatch
 from django.urls.resolvers import URLPattern
-from rest_framework.decorators import action, api_view
+from rest_framework.decorators import action, api_view, permission_classes
+from rest_framework.permissions import IsAuthenticatedOrReadOnly
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet
 
+from dispatcher.utils import resolve_callable
+
 from ansible_base.lib.utils.response import get_fully_qualified_url
 from ansible_base.lib.utils.views.ansible_base import AnsibleBaseView
 from ansible_base.oauth2_provider.permissions import OAuth2ScopePermission
@@ -177,6 +180,7 @@ def api_root(request, format=None):
     list_endpoints['service-index'] = get_fully_qualified_url('service-index-root')
     list_endpoints['role-metadata'] = get_fully_qualified_url('role-metadata')
     list_endpoints['timeout-view'] = get_fully_qualified_url('test-timeout-view')
+    list_endpoints['trigger-task'] = get_fully_qualified_url('test-trigger-task')
 
     return Response(list_endpoints)
 
@@ -186,6 +190,20 @@ def timeout_view(request, format=None):
     time.sleep(60 * 10)  # 10 minutes
 
 
+@api_view(['GET', 'POST'])
+@permission_classes([IsAuthenticatedOrReadOnly])
+def trigger_task(request, format=None):
+    if request.method == 'GET':
+        return Response({})
+    elif request.method == 'POST':
+        try:
+            fn = resolve_callable(request.data)
+        except Exception as exc:
+            return Response({'error': f'Could not locate task {request.data}, error: {str(exc)}'})
+        fn.delay()
+        return Response({'task': request.data})
+
+
 class MultipleFieldsViewSet(TestAppViewSet):
     serializer_class = serializers.MultipleFieldsModelSerializer