Set up export/import action mixin

docs/api_drf.rst

@@ -14,9 +14,27 @@ API (Rest Framework)
 .. autoclass:: import_export_extensions.api.ExportJobForUserViewSet
    :members:
 
+.. autoclass:: import_export_extensions.api.BaseImportJobViewSet
+   :members:
+
+.. autoclass:: import_export_extensions.api.BaseExportJobViewSet
+   :members:
+
+.. autoclass:: import_export_extensions.api.BaseImportJobForUserViewSet
+   :members:
+
+.. autoclass:: import_export_extensions.api.BaseExportJobForUserViewSet
+   :members:
+
 .. autoclass:: import_export_extensions.api.LimitQuerySetToCurrentUserMixin
    :members:
 
+.. autoclass:: import_export_extensions.api.ImportStartActionMixin
+   :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 ability 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.py → import_export_extensions/api/mixins/common.py

@@ -3,7 +3,10 @@ class LimitQuerySetToCurrentUserMixin:
 
     def get_queryset(self):
         """Return user's jobs."""
-        if self.action == "start":
+        if self.action in (
+            getattr(self, "import_action", ""),
+            getattr(self, "export_action", ""),
+        ):
             # To make it consistent and for better support of drf-spectacular
             return super().get_queryset()  # pragma: no cover
         return (

import_export_extensions/api/mixins/export_mixins.py

@@ -0,0 +1,141 @@
+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_action"
+    export_action_name = "export"
+    export_action_url = "export"
+    export_detail_serializer_class = serializers.ExportJobSerializer
+    export_ordering: collections.abc.Sequence[str] = ()
+    export_ordering_fields: collections.abc.Sequence[str] = ()
+    export_open_api_description = (
+        "This endpoint creates export job and starts it. "
+        "To monitor progress use detail endpoint for jobs to fetch state of "
+        "job. Once it's status is `EXPORTED`, you can download file."
+    )
+
+    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,
+                ),
+            )
+
+        def start_export_action(
+            self: "ExportStartActionMixin",
+            request: request.Request,
+        ) -> response.Response:
+            return self.start_export(request)
+
+        setattr(cls, cls.export_action, start_export_action)
+        decorators.action(
+            methods=["POST"],
+            url_name=cls.export_action_name,
+            url_path=cls.export_action_url,
+            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(
+                        description=cls.export_open_api_description,
+                        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:
+        """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,115 @@
+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_action"
+    import_action_name = "import"
+    import_action_url = "import"
+    import_detail_serializer_class = serializers.ImportJobSerializer
+    import_open_api_description = (
+        "This endpoint creates import job and starts it. "
+        "To monitor progress use detail endpoint for jobs to fetch state of "
+        "job. Once it's status is `PARSED`, you can confirm import and data "
+        "should start importing. When status `INPUT_ERROR` or `PARSE_ERROR` "
+        "it means data failed validations and can't be imported. "
+        "When status is `IMPORTED`, it means data is in system and "
+        "job is completed."
+    )
+
+    def __init_subclass__(cls) -> None:
+        super().__init_subclass__()
+        # Skip if it is has no resource_class specified
+        if not hasattr(cls, "resource_class"):
+            return
+
+        def start_import_action(
+            self: "ImportStartActionMixin",
+            request: request.Request,
+        ) -> response.Response:
+            return self.start_import(request)
+
+        setattr(cls, cls.import_action, start_import_action)
+        decorators.action(
+            methods=["POST"],
+            url_name=cls.import_action_name,
+            url_path=cls.import_action_url,
+            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(
+                        description=cls.import_open_api_description,
+                        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)
+
+    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]