Set up export/import action mixin

docs/api_drf.rst

@@ -14,9 +14,18 @@ API (Rest Framework)
 .. autoclass:: import_export_extensions.api.ExportJobForUserViewSet
    :members:
 
+.. autoclass:: import_export_extensions.api.BaseExportJobViewSet
+   :members:
+
+.. autoclass:: import_export_extensions.api.BaseExportJobForUserViewSet
+   :members:
+
 .. autoclass:: import_export_extensions.api.LimitQuerySetToCurrentUserMixin
    :members:
 
+.. autoclass:: import_export_extensions.api.ExportStartActionMixin
+   :members:
+
 .. autoclass:: import_export_extensions.api.CreateExportJob
    :members: create, validate
 

docs/getting_started.rst

@@ -169,3 +169,17 @@ the OpenAPI specification will be available.
 .. figure:: _static/images/bands-openapi.png
 
    A screenshot of the generated OpenAPI specification
+
+
+Import/Export API actions mixins
+-----------------
+
+Alternatively you can use ``api.mixins.ExportStartActionMixin`` and ``api.mixins.ImportStartActionMixin``
+to add to your current viewsets to create import/export jobs.
+You would also need to use ``api.views.BaseExportJobViewSet/BaseExportJobForUsersViewSet``
+and ``api.views.BaseExportJobViewSet/BaseImportJobForUsersViewSet`` to setup endpoints to be able to:
+
+* ``list`` - Returns a list of jobs for the ``resource_class`` set in ViewSet
+* ``retrieve`` - Returns details of a job based on the provided ID
+* ``cancel`` - Stops the import/export process and sets the job's status to ``CANCELLED``.
+* ``confirm`` - Confirms the import after the parse stage. This action is available only in import jobs.

import_export_extensions/api/__init__.py

@@ -1,4 +1,8 @@
-from .mixins import LimitQuerySetToCurrentUserMixin
+from .mixins import (
+    ExportStartActionMixin,
+    ImportStartActionMixin,
+    LimitQuerySetToCurrentUserMixin,
+)
 from .serializers import (
     CreateExportJob,
     CreateImportJob,
@@ -8,6 +12,10 @@
     ProgressSerializer,
 )
 from .views import (
+    BaseExportJobForUserViewSet,
+    BaseExportJobViewSet,
+    BaseImportJobForUserViewSet,
+    BaseImportJobViewSet,
     ExportJobForUserViewSet,
     ExportJobViewSet,
     ImportJobForUserViewSet,

import_export_extensions/api/mixins/__init__.py

@@ -0,0 +1,3 @@
+from .common import LimitQuerySetToCurrentUserMixin
+from .export_mixins import ExportStartActionMixin
+from .import_mixins import ImportStartActionMixin

import_export_extensions/api/mixins/export_mixins.py

@@ -0,0 +1,129 @@
+import collections.abc
+import contextlib
+import typing
+
+from django.conf import settings
+from django.utils import module_loading
+
+from rest_framework import (
+    decorators,
+    request,
+    response,
+    status,
+)
+
+from ... import resources
+from .. import serializers
+
+
+class ExportStartActionMixin:
+    """Mixin which adds start export action."""
+
+    resource_class: type[resources.CeleryModelResource]
+    export_action = "start_export"
+    export_detail_serializer_class = serializers.ExportJobSerializer
+    export_ordering: collections.abc.Sequence[str] = ()
+    export_ordering_fields: collections.abc.Sequence[str] = ()
+
+    def __init_subclass__(cls) -> None:
+        super().__init_subclass__()
+        # Skip if it is has no resource_class specified
+        if not hasattr(cls, "resource_class"):
+            return
+        filter_backends = [
+            module_loading.import_string(
+                settings.DRF_EXPORT_DJANGO_FILTERS_BACKEND,
+            ),
+        ]
+        if cls.export_ordering_fields:
+            filter_backends.append(
+                module_loading.import_string(
+                    settings.DRF_EXPORT_ORDERING_BACKEND,
+                ),
+            )
+        decorators.action(
+            methods=["POST"],
+            url_name=cls.export_action.replace("_", "-"),
+            url_path=cls.export_action.replace("_", "-"),
+            detail=False,
+            queryset=cls.resource_class.get_model_queryset(),
+            serializer_class=cls().get_export_create_serializer_class(),
+            filterset_class=getattr(
+                cls.resource_class,
+                "filterset_class",
+                None,
+            ),
+            filter_backends=filter_backends,
+            ordering=cls.export_ordering,
+            ordering_fields=cls.export_ordering_fields,
+        )(getattr(cls, cls.export_action))
+        # Correct specs of drf-spectacular if it is installed
+        with contextlib.suppress(ImportError):
+            from drf_spectacular import utils
+
+            utils.extend_schema_view(
+                **{
+                    cls.export_action: utils.extend_schema(
+                        filters=True,
+                        responses={
+                            status.HTTP_201_CREATED: cls().get_export_detail_serializer_class(),  # noqa: E501
+                        },
+                    ),
+                },
+            )(cls)
+
+    def get_queryset(self):
+        """Return export model queryset on export action.
+
+        For better openapi support and consistency.
+
+        """
+        if self.action == self.export_action:
+            return self.resource_class.get_model_queryset()  # pragma: no cover
+        return super().get_queryset()
+
+    def get_export_detail_serializer_class(self):
+        """Get serializer which will be used show details of export job."""
+        return self.export_detail_serializer_class
+
+    def get_export_create_serializer_class(self):
+        """Get serializer which will be used to start export job."""
+        return serializers.get_create_export_job_serializer(
+            self.resource_class,
+        )
+
+    def get_export_resource_kwargs(self) -> dict[str, typing.Any]:
+        """Provide extra arguments to resource class."""
+        return {}
+
+    def get_serializer(self, *args, **kwargs):
+        """Provide resource kwargs to serializer class."""
+        if self.action == self.export_action:
+            kwargs.setdefault(
+                "resource_kwargs",
+                self.get_export_resource_kwargs(),
+            )
+        return super().get_serializer(*args, **kwargs)
+
+    def start_export(self, request: request.Request) -> response.Response:
+        """Start export job."""
+        return self._start_export(request)
+
+    def _start_export(self, request: request.Request) -> response.Response:
+        """Validate request data and start ExportJob."""
+        ordering = request.query_params.get("ordering", "")
+        if ordering:
+            ordering = ordering.split(",")
+        serializer = self.get_serializer(
+            data=request.data,
+            ordering=ordering,
+            filter_kwargs=request.query_params,
+        )
+        serializer.is_valid(raise_exception=True)
+        export_job = serializer.save()
+        return response.Response(
+            data=self.get_export_detail_serializer_class()(
+                instance=export_job,
+            ).data,
+            status=status.HTTP_201_CREATED,
+        )

import_export_extensions/api/mixins/import_mixins.py

@@ -0,0 +1,100 @@
+import contextlib
+import typing
+
+from rest_framework import (
+    decorators,
+    request,
+    response,
+    status,
+)
+
+from ... import resources
+from .. import serializers
+
+
+class ImportStartActionMixin:
+    """Mixin which adds start import action."""
+
+    resource_class: type[resources.CeleryModelResource]
+    import_action = "start_import"
+    import_detail_serializer_class = serializers.ImportJobSerializer
+
+    def __init_subclass__(cls) -> None:
+        super().__init_subclass__()
+        # Skip if it is has no resource_class specified
+        if not hasattr(cls, "resource_class"):
+            return
+        decorators.action(
+            methods=["POST"],
+            url_name=cls.import_action.replace("_", "-"),
+            url_path=cls.import_action.replace("_", "-"),
+            detail=False,
+            queryset=cls.resource_class.get_model_queryset(),
+            serializer_class=cls().get_import_create_serializer_class(),
+        )(getattr(cls, cls.import_action))
+        # Correct specs of drf-spectacular if it is installed
+        with contextlib.suppress(ImportError):
+            from drf_spectacular import utils
+
+            utils.extend_schema_view(
+                **{
+                    cls.import_action: utils.extend_schema(
+                        filters=True,
+                        responses={
+                            status.HTTP_201_CREATED: cls().get_import_detail_serializer_class(),  # noqa: E501
+                        },
+                    ),
+                },
+            )(cls)
+
+    def get_queryset(self):
+        """Return import model queryset on import action.
+
+        For better openapi support and consistency.
+
+        """
+        if self.action == self.import_action:
+            return self.resource_class.get_model_queryset()  # pragma: no cover
+        return super().get_queryset()
+
+    def get_import_detail_serializer_class(self):
+        """Get serializer which will be used show details of import job."""
+        return self.import_detail_serializer_class
+
+    def get_import_create_serializer_class(self):
+        """Get serializer which will be used to start import job."""
+        return serializers.get_create_import_job_serializer(
+            self.resource_class,
+        )
+
+    def get_import_resource_kwargs(self) -> dict[str, typing.Any]:
+        """Provide extra arguments to resource class."""
+        return {}
+
+    def get_serializer(self, *args, **kwargs):
+        """Provide resource kwargs to serializer class."""
+        if self.action == self.import_action:
+            kwargs.setdefault(
+                "resource_kwargs",
+                self.get_import_resource_kwargs(),
+            )
+        return super().get_serializer(*args, **kwargs)
+
+    # Can't use import `word``
+    def start_import(self, request: request.Request) -> response.Response:
+        """Start import job."""
+        return self._start_import(request)
+
+    def _start_import(self, request: request.Request) -> response.Response:
+        """Validate request data and start ImportJob."""
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        import_job = serializer.save()
+
+        return response.Response(
+            data=self.get_import_detail_serializer_class()(
+                instance=import_job,
+            ).data,
+            status=status.HTTP_201_CREATED,
+        )

import_export_extensions/api/serializers/export_job.py

@@ -99,10 +99,19 @@ def update(self, instance, validated_data):
         """Empty method to pass linters checks."""
 
 
+# Use it to cache already generated serializers to avoid duplication
+_GENERATED_EXPORT_JOB_SERIALIZERS: dict[
+    type[resources.CeleryModelResource],
+    type,
+] = {}
+
+
 def get_create_export_job_serializer(
     resource: type[resources.CeleryModelResource],
 ) -> type:
     """Create serializer for ExportJobs creation."""
+    if resource in _GENERATED_EXPORT_JOB_SERIALIZERS:
+        return _GENERATED_EXPORT_JOB_SERIALIZERS[resource]
 
     class _CreateExportJob(CreateExportJob):
         """Serializer to start export job."""
@@ -116,8 +125,9 @@ class _CreateExportJob(CreateExportJob):
             ],
         )
 
-    return type(
+    _GENERATED_EXPORT_JOB_SERIALIZERS[resource] = type(
         f"{resource.__name__}CreateExportJob",
         (_CreateExportJob,),
         {},
     )
+    return _GENERATED_EXPORT_JOB_SERIALIZERS[resource]

import_export_extensions/api/serializers/import_job.py

@@ -126,18 +126,28 @@ def update(self, instance, validated_data):
         """Empty method to pass linters checks."""
 
 
+# Use it to cache already generated serializers to avoid duplication
+_GENERATED_IMPORT_JOB_SERIALIZERS: dict[
+    type[resources.CeleryModelResource],
+    type,
+] = {}
+
+
 def get_create_import_job_serializer(
     resource: type[resources.CeleryModelResource],
 ) -> type:
     """Create serializer for ImportJobs creation."""
+    if resource in _GENERATED_IMPORT_JOB_SERIALIZERS:
+        return _GENERATED_IMPORT_JOB_SERIALIZERS[resource]
 
     class _CreateImportJob(CreateImportJob):
         """Serializer to start import job."""
 
         resource_class: type[resources.CeleryModelResource] = resource
 
-    return type(
+    _GENERATED_IMPORT_JOB_SERIALIZERS[resource] = type(
         f"{resource.__name__}CreateImportJob",
         (_CreateImportJob,),
         {},
     )
+    return _GENERATED_IMPORT_JOB_SERIALIZERS[resource]

import_export_extensions/api/views/__init__.py

@@ -1,8 +1,12 @@
 from .export_job import (
+    BaseExportJobForUserViewSet,
+    BaseExportJobViewSet,
     ExportJobForUserViewSet,
     ExportJobViewSet,
 )
 from .import_job import (
+    BaseImportJobForUserViewSet,
+    BaseImportJobViewSet,
     ImportJobForUserViewSet,
     ImportJobViewSet,
 )