Enabled dispatching customization for collections

https://pulp.plan.io/issues/6673
closes #6673

CHANGES/6673.misc

@@ -0,0 +1 @@
+Enabled dispatching customization for importing collections

pulp_ansible/app/galaxy/mixins.py

@@ -0,0 +1,21 @@
+from pulpcore.plugin.tasking import enqueue_with_reservation
+
+from pulp_ansible.app.tasks.collections import import_collection
+
+
+class UploadGalaxyCollectionMixin:
+    """
+    Provides a method that dispatches a task.
+    """
+
+    def _dispatch_import_collection_task(self, artifact_pk, repository=None, **kwargs):
+        """
+        Dispatch a Import Collection creation task.
+        """
+        locks = [str(artifact_pk)]
+        kwargs["artifact_pk"] = artifact_pk
+        if repository:
+            locks.append(repository)
+            kwargs["repository_pk"] = repository.pk
+
+        return enqueue_with_reservation(import_collection, locks, kwargs=kwargs)

pulp_ansible/app/galaxy/v3/views.py

@@ -17,7 +17,6 @@
 from pulpcore.plugin.exceptions import DigestValidationError
 from pulpcore.plugin.models import Artifact, Content, ContentArtifact
 from pulpcore.plugin.serializers import AsyncOperationResponseSerializer
-from pulpcore.plugin.tasking import enqueue_with_reservation
 from rest_framework.reverse import reverse
 
 from pulp_ansible.app.galaxy.v3.exceptions import ExceptionHandlerMixin
@@ -31,8 +30,8 @@
     CollectionOneShotSerializer,
     CollectionImportDetailSerializer,
 )
-from pulp_ansible.app.tasks.collections import import_collection
 
+from pulp_ansible.app.galaxy.mixins import UploadGalaxyCollectionMixin
 from pulp_ansible.app.viewsets import CollectionVersionFilter
 
 
@@ -109,7 +108,9 @@ def update(self, request, *args, **kwargs):
         return Response(serializer.data)
 
 
-class CollectionUploadViewSet(ExceptionHandlerMixin, viewsets.GenericViewSet):
+class CollectionUploadViewSet(
+    ExceptionHandlerMixin, viewsets.GenericViewSet, UploadGalaxyCollectionMixin
+):
     """
     ViewSet for Collection Uploads.
     """
@@ -150,8 +151,7 @@ def create(self, request, path):
         except IntegrityError:
             raise serializers.ValidationError(_("Artifact already exists."))
 
-        locks = [str(artifact.pk)]
-        kwargs = {"artifact_pk": artifact.pk}
+        kwargs = {}
 
         if serializer.validated_data["expected_namespace"]:
             kwargs["expected_namespace"] = serializer.validated_data["expected_namespace"]
@@ -162,11 +162,9 @@ def create(self, request, path):
         if serializer.validated_data["expected_version"]:
             kwargs["expected_version"] = serializer.validated_data["expected_version"]
 
-        if distro.repository:
-            locks.append(distro.repository)
-            kwargs["repository_pk"] = distro.repository.pk
-
-        async_result = enqueue_with_reservation(import_collection, locks, kwargs=kwargs)
+        async_result = self._dispatch_import_collection_task(
+            artifact.pk, distro.repository, **kwargs
+        )
         CollectionImport.objects.create(task_id=async_result.id)
 
         data = {

pulp_ansible/app/galaxy/views.py

@@ -6,10 +6,9 @@
 
 from pulpcore.app.models import Artifact
 from pulpcore.app.response import OperationPostponedResponse
-from pulpcore.tasking.tasks import enqueue_with_reservation
 from pulpcore.plugin.models import ContentArtifact
 
-from pulp_ansible.app.tasks.collections import import_collection
+from pulp_ansible.app.galaxy.mixins import UploadGalaxyCollectionMixin
 from pulp_ansible.app.models import AnsibleDistribution, Collection, CollectionVersion, Role
 
 from .serializers import (
@@ -116,7 +115,7 @@ def get(self, request, path=None, namespace=None, name=None):
         return response.Response(GalaxyCollectionSerializer(collection).data)
 
 
-class GalaxyCollectionView(generics.ListAPIView):
+class GalaxyCollectionView(generics.ListAPIView, UploadGalaxyCollectionMixin):
     """
     View for Collection models.
     """
@@ -157,13 +156,7 @@ def post(self, request, path):
         artifact = Artifact.init_and_validate(serializer.validated_data["file"])
         artifact.save()
 
-        locks = [str(artifact.pk)]
-        kwargs = {"artifact_pk": artifact.pk}
-        if distro.repository:
-            locks.append(distro.repository)
-            kwargs["repository_pk"] = distro.repository.pk
-
-        async_result = enqueue_with_reservation(import_collection, locks, kwargs=kwargs)
+        async_result = self._dispatch_import_collection_task(artifact.pk, distro.repository)
         return OperationPostponedResponse(async_result, request)
 
 

pulp_ansible/app/viewsets.py

@@ -33,6 +33,7 @@
     RepositoryViewSet,
     RepositoryVersionViewSet,
 )
+from pulp_ansible.app.galaxy.mixins import UploadGalaxyCollectionMixin
 from .models import (
     AnsibleDistribution,
     AnsibleRemote,
@@ -55,7 +56,6 @@
     TagSerializer,
 )
 from .tasks.collections import sync as collection_sync
-from .tasks.collections import import_collection
 from .tasks.synchronizing import synchronize as role_sync
 
 
@@ -276,7 +276,7 @@ class CollectionRemoteViewSet(RemoteViewSet):
     serializer_class = CollectionRemoteSerializer
 
 
-class CollectionUploadViewSet(viewsets.ViewSet):
+class CollectionUploadViewSet(viewsets.ViewSet, UploadGalaxyCollectionMixin):
     """
     ViewSet for One Shot Collection Upload.
 
@@ -320,9 +320,7 @@ def create(self, request):
         except IntegrityError:
             raise serializers.ValidationError(_("Artifact already exists."))
 
-        async_result = enqueue_with_reservation(
-            import_collection, [str(artifact.pk)], kwargs={"artifact_pk": artifact.pk}
-        )
+        async_result = self._dispatch_import_collection_task(artifact.pk)
 
         return OperationPostponedResponse(async_result, request)