Skip to content

API (Rest Framework)

Bases: ImportStartActionMixin, BaseImportJobViewSet

Base API viewset for ImportJob model.

Based on resource_class it will generate an endpoint which will allow to start an import to model which was specified in resource_class. On success this endpoint we return an instance of import.

Endpoints
  • list - to get list of all import jobs
  • details(retrieve) - to get status of import job
  • start - create import job and start parsing data from attached file
  • confirm - confirm import after parsing process is finished
  • cancel - stop importing/parsing process and cancel this import job
Source code in import_export_extensions/api/views/import_job.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class ImportJobViewSet(
    core_mixins.ImportStartActionMixin,
    BaseImportJobViewSet,
):
    """Base API viewset for ImportJob model.

    Based on resource_class it will generate an endpoint which will allow to
    start an import to model which was specified in resource_class. On success
    this endpoint we return an instance of import.

    Endpoints:
        - list - to get list of all import jobs
        - details(retrieve) - to get status of import job
        - start - create import job and start parsing data from attached file
        - confirm - confirm import after parsing process is finished
        - cancel - stop importing/parsing process and cancel this import job
    """

    import_action_name = "start"
    import_action_url = "start"

    def get_queryset(self) -> QuerySet[models.ImportJob]:
        """Filter import jobs by resource used in viewset."""
        if self.action == getattr(self, "import_action", ""):
            # To make it consistent and for better support of drf-spectacular
            return super().get_queryset()  # pragma: no cover
        return super().get_queryset().filter(
            resource_path=self.resource_class.class_path,
        )

get_queryset()

Filter import jobs by resource used in viewset.

Source code in import_export_extensions/api/views/import_job.py 
129
130
131
132
133
134
135
136
def get_queryset(self) -> QuerySet[models.ImportJob]:
    """Filter import jobs by resource used in viewset."""
    if self.action == getattr(self, "import_action", ""):
        # To make it consistent and for better support of drf-spectacular
        return super().get_queryset()  # pragma: no cover
    return super().get_queryset().filter(
        resource_path=self.resource_class.class_path,
    )

Bases: ExportStartActionMixin, BaseExportJobViewSet

Base API viewset for ExportJob model.

Based on resource_class it will generate an endpoint which will allow to start an export of model which was specified in resource_class. This endpoint will support filtration based on FilterSet class specified in resource. On success this endpoint we return an instance of export, to get status of job, just use detail(retrieve) endpoint.

Source code in import_export_extensions/api/views/export_job.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class ExportJobViewSet(
    core_mixins.ExportStartActionMixin,
    BaseExportJobViewSet,
):
    """Base API viewset for ExportJob model.

    Based on resource_class it will generate an endpoint which will allow to
    start an export of model which was specified in resource_class. This
    endpoint will support filtration based on FilterSet class specified in
    resource. On success this endpoint we return an instance of export, to
    get status of job, just use detail(retrieve) endpoint.

    """

    export_action_name = "start"
    export_action_url = "start"

    def get_queryset(self) -> QuerySet[models.ExportJob]:
        """Filter export jobs by resource used in viewset."""
        if self.action == getattr(self, "export_action", ""):
            # To make it consistent and for better support of drf-spectacular
            return super().get_queryset()  # pragma: no cover
        return super().get_queryset().filter(
            resource_path=self.resource_class.class_path,
        )

get_queryset()

Filter export jobs by resource used in viewset.

Source code in import_export_extensions/api/views/export_job.py 
103
104
105
106
107
108
109
110
def get_queryset(self) -> QuerySet[models.ExportJob]:
    """Filter export jobs by resource used in viewset."""
    if self.action == getattr(self, "export_action", ""):
        # To make it consistent and for better support of drf-spectacular
        return super().get_queryset()  # pragma: no cover
    return super().get_queryset().filter(
        resource_path=self.resource_class.class_path,
    )

Bases: LimitQuerySetToCurrentUserMixin, ImportJobViewSet

Viewset for providing import feature to users.

Source code in import_export_extensions/api/views/import_job.py 
138
139
140
141
142
class ImportJobForUserViewSet(
    core_mixins.LimitQuerySetToCurrentUserMixin,
    ImportJobViewSet,
):
    """Viewset for providing import feature to users."""

Bases: LimitQuerySetToCurrentUserMixin, ExportJobViewSet

Viewset for providing export feature to users.

Source code in import_export_extensions/api/views/export_job.py 
113
114
115
116
117
class ExportJobForUserViewSet(
    core_mixins.LimitQuerySetToCurrentUserMixin,
    ExportJobViewSet,
):
    """Viewset for providing export feature to users."""

Bases: ListModelMixin, RetrieveModelMixin, GenericViewSet

Base viewset for managing import jobs.

Source code in import_export_extensions/api/views/import_job.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class BaseImportJobViewSet(
    mixins.ListModelMixin,
    mixins.RetrieveModelMixin,
    viewsets.GenericViewSet,
):
    """Base viewset for managing import jobs."""

    permission_classes = (permissions.IsAuthenticated,)
    serializer_class = core_mixins.ImportStartActionMixin.import_detail_serializer_class  # noqa: E501
    queryset = models.ImportJob.objects.all()
    search_fields: collections.abc.Sequence[str] = ("id",)
    ordering: collections.abc.Sequence[str] = (
        "id",
    )
    ordering_fields: collections.abc.Sequence[str] = (
        "id",
        "created",
        "modified",
    )

    def __init_subclass__(cls) -> None:
        """Dynamically create an cancel api endpoints.

        Need to do this to enable action and correct open-api spec generated by
        drf_spectacular.

        """
        super().__init_subclass__()
        decorators.action(
            methods=["POST"],
            detail=True,
        )(cls.cancel)
        decorators.action(
            methods=["POST"],
            detail=True,
        )(cls.confirm)
        # Correct specs of drf-spectacular if it is installed
        with contextlib.suppress(ImportError):
            from drf_spectacular.utils import extend_schema, extend_schema_view
            if hasattr(cls, "get_import_detail_serializer_class"):
                response_serializer = cls().get_import_detail_serializer_class()  # noqa: E501
            else:
                response_serializer = cls().get_serializer_class()
            extend_schema_view(
                cancel=extend_schema(
                    request=None,
                    responses={
                        status.HTTP_200_OK: response_serializer,
                    },
                ),
                confirm=extend_schema(
                    request=None,
                    responses={
                        status.HTTP_200_OK: response_serializer,
                    },
                ),
            )(cls)

    def confirm(self, *args, **kwargs) -> response.Response:
        """Confirm import job that has `parsed` status."""
        job: models.ImportJob = self.get_object()

        try:
            job.confirm_import()
        except ValueError as error:
            raise exceptions.ValidationError(error.args[0]) from error

        serializer = self.get_serializer(instance=job)
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializer.data,
        )

    def cancel(self, *args, **kwargs) -> response.Response:
        """Cancel import job that is in progress."""
        job: models.ImportJob = self.get_object()

        try:
            job.cancel_import()
        except ValueError as error:
            raise exceptions.ValidationError(error.args[0]) from error

        serializer = self.get_serializer(instance=job)
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializer.data,
        )

__init_subclass__()

Dynamically create an cancel api endpoints.

Need to do this to enable action and correct open-api spec generated by drf_spectacular.

Source code in import_export_extensions/api/views/import_job.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init_subclass__(cls) -> None:
    """Dynamically create an cancel api endpoints.

    Need to do this to enable action and correct open-api spec generated by
    drf_spectacular.

    """
    super().__init_subclass__()
    decorators.action(
        methods=["POST"],
        detail=True,
    )(cls.cancel)
    decorators.action(
        methods=["POST"],
        detail=True,
    )(cls.confirm)
    # Correct specs of drf-spectacular if it is installed
    with contextlib.suppress(ImportError):
        from drf_spectacular.utils import extend_schema, extend_schema_view
        if hasattr(cls, "get_import_detail_serializer_class"):
            response_serializer = cls().get_import_detail_serializer_class()  # noqa: E501
        else:
            response_serializer = cls().get_serializer_class()
        extend_schema_view(
            cancel=extend_schema(
                request=None,
                responses={
                    status.HTTP_200_OK: response_serializer,
                },
            ),
            confirm=extend_schema(
                request=None,
                responses={
                    status.HTTP_200_OK: response_serializer,
                },
            ),
        )(cls)

cancel(*args, **kwargs)

Cancel import job that is in progress.

Source code in import_export_extensions/api/views/import_job.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def cancel(self, *args, **kwargs) -> response.Response:
    """Cancel import job that is in progress."""
    job: models.ImportJob = self.get_object()

    try:
        job.cancel_import()
    except ValueError as error:
        raise exceptions.ValidationError(error.args[0]) from error

    serializer = self.get_serializer(instance=job)
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializer.data,
    )

confirm(*args, **kwargs)

Confirm import job that has parsed status.

Source code in import_export_extensions/api/views/import_job.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def confirm(self, *args, **kwargs) -> response.Response:
    """Confirm import job that has `parsed` status."""
    job: models.ImportJob = self.get_object()

    try:
        job.confirm_import()
    except ValueError as error:
        raise exceptions.ValidationError(error.args[0]) from error

    serializer = self.get_serializer(instance=job)
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializer.data,
    )

Bases: ListModelMixin, RetrieveModelMixin, GenericViewSet

Base viewset for managing export jobs.

Source code in import_export_extensions/api/views/export_job.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class BaseExportJobViewSet(
    mixins.ListModelMixin,
    mixins.RetrieveModelMixin,
    viewsets.GenericViewSet,
):
    """Base viewset for managing export jobs."""

    permission_classes = (permissions.IsAuthenticated,)
    serializer_class = core_mixins.ExportStartActionMixin.export_detail_serializer_class  # noqa: E501
    queryset = models.ExportJob.objects.all()
    filterset_class: django_filters.rest_framework.FilterSet | None = None
    search_fields: collections.abc.Sequence[str] = ("id",)
    ordering: collections.abc.Sequence[str] = (
        "id",
    )
    ordering_fields: collections.abc.Sequence[str] = (
        "id",
        "created",
        "modified",
    )

    def __init_subclass__(cls) -> None:
        """Dynamically create an cancel api endpoints.

        Need to do this to enable action and correct open-api spec generated by
        drf_spectacular.

        """
        super().__init_subclass__()
        decorators.action(
            methods=["POST"],
            detail=True,
        )(cls.cancel)
        # Correct specs of drf-spectacular if it is installed
        with contextlib.suppress(ImportError):
            from drf_spectacular.utils import extend_schema, extend_schema_view
            if hasattr(cls, "get_export_detail_serializer_class"):
                response_serializer = cls().get_export_detail_serializer_class()  # noqa: E501
            else:
                response_serializer = cls().get_serializer_class()
            extend_schema_view(
                cancel=extend_schema(
                    request=None,
                    responses={
                        status.HTTP_200_OK: response_serializer,
                    },
                ),
            )(cls)

    def cancel(self, *args, **kwargs) -> response.Response:
        """Cancel export job that is in progress."""
        job: models.ExportJob = self.get_object()

        try:
            job.cancel_export()
        except ValueError as error:
            raise exceptions.ValidationError(error.args[0]) from error

        serializer = self.get_serializer(instance=job)
        return response.Response(
            status=status.HTTP_200_OK,
            data=serializer.data,
        )

__init_subclass__()

Dynamically create an cancel api endpoints.

Need to do this to enable action and correct open-api spec generated by drf_spectacular.

Source code in import_export_extensions/api/views/export_job.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init_subclass__(cls) -> None:
    """Dynamically create an cancel api endpoints.

    Need to do this to enable action and correct open-api spec generated by
    drf_spectacular.

    """
    super().__init_subclass__()
    decorators.action(
        methods=["POST"],
        detail=True,
    )(cls.cancel)
    # Correct specs of drf-spectacular if it is installed
    with contextlib.suppress(ImportError):
        from drf_spectacular.utils import extend_schema, extend_schema_view
        if hasattr(cls, "get_export_detail_serializer_class"):
            response_serializer = cls().get_export_detail_serializer_class()  # noqa: E501
        else:
            response_serializer = cls().get_serializer_class()
        extend_schema_view(
            cancel=extend_schema(
                request=None,
                responses={
                    status.HTTP_200_OK: response_serializer,
                },
            ),
        )(cls)

cancel(*args, **kwargs)

Cancel export job that is in progress.

Source code in import_export_extensions/api/views/export_job.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def cancel(self, *args, **kwargs) -> response.Response:
    """Cancel export job that is in progress."""
    job: models.ExportJob = self.get_object()

    try:
        job.cancel_export()
    except ValueError as error:
        raise exceptions.ValidationError(error.args[0]) from error

    serializer = self.get_serializer(instance=job)
    return response.Response(
        status=status.HTTP_200_OK,
        data=serializer.data,
    )

Bases: LimitQuerySetToCurrentUserMixin, BaseImportJobViewSet

Viewset for providing export job management to users.

Source code in import_export_extensions/api/views/import_job.py 
144
145
146
147
148
class BaseImportJobForUserViewSet(
    core_mixins.LimitQuerySetToCurrentUserMixin,
    BaseImportJobViewSet,
):
    """Viewset for providing export job management to users."""

Bases: LimitQuerySetToCurrentUserMixin, BaseExportJobViewSet

Viewset for providing export job management to users.

Source code in import_export_extensions/api/views/export_job.py 
119
120
121
122
123
class BaseExportJobForUserViewSet(
    core_mixins.LimitQuerySetToCurrentUserMixin,
    BaseExportJobViewSet,
):
    """Viewset for providing export job management to users."""

Make queryset to return only current user jobs.

Source code in import_export_extensions/api/mixins/common.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class LimitQuerySetToCurrentUserMixin:
    """Make queryset to return only current user jobs."""

    def get_queryset(self) -> QuerySet:
        """Return user's jobs."""
        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()  # type: ignore[misc]
            )  # pragma: no cover
        return (
            super()  # type: ignore[misc]
            .get_queryset()
            .filter(created_by_id=getattr(self.request.user, "pk", None))
        )

get_queryset()

Return user's jobs.

Source code in import_export_extensions/api/mixins/common.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def get_queryset(self) -> QuerySet:
    """Return user's jobs."""
    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()  # type: ignore[misc]
        )  # pragma: no cover
    return (
        super()  # type: ignore[misc]
        .get_queryset()
        .filter(created_by_id=getattr(self.request.user, "pk", None))
    )

Mixin which adds start import action.

Source code in import_export_extensions/api/mixins/import_mixins.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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,
            *args,
            **kwargs,
        ) -> 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) -> QuerySet:
        """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()  # type: ignore[misc]

    def get_import_detail_serializer_class(
        self,
    ) -> type[serializers.ImportJobSerializer]:
        """Get serializer which will be used show details of import job."""
        return self.import_detail_serializer_class

    def get_import_create_serializer_class(self) -> type:
        """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) -> Serializer:
        """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)  # type: ignore[misc]

    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,
        )

get_import_create_serializer_class()

Get serializer which will be used to start import job.

Source code in import_export_extensions/api/mixins/import_mixins.py 
91
92
93
94
95
def get_import_create_serializer_class(self) -> type:
    """Get serializer which will be used to start import job."""
    return serializers.get_create_import_job_serializer(
        self.resource_class,
    )

get_import_detail_serializer_class()

Get serializer which will be used show details of import job.

Source code in import_export_extensions/api/mixins/import_mixins.py 
85
86
87
88
89
def get_import_detail_serializer_class(
    self,
) -> type[serializers.ImportJobSerializer]:
    """Get serializer which will be used show details of import job."""
    return self.import_detail_serializer_class

get_import_resource_kwargs()

Provide extra arguments to resource class.

Source code in import_export_extensions/api/mixins/import_mixins.py 
97
98
99
def get_import_resource_kwargs(self) -> dict[str, typing.Any]:
    """Provide extra arguments to resource class."""
    return {}

get_queryset()

Return import model queryset on import action.

For better openapi support and consistency.

Source code in import_export_extensions/api/mixins/import_mixins.py 
75
76
77
78
79
80
81
82
83
def get_queryset(self) -> QuerySet:
    """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()  # type: ignore[misc]

get_serializer(*args, **kwargs)

Provide resource kwargs to serializer class.

Source code in import_export_extensions/api/mixins/import_mixins.py 
101
102
103
104
105
106
107
108
def get_serializer(self, *args, **kwargs) -> Serializer:
    """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)  # type: ignore[misc]

start_import(request)

Validate request data and start ImportJob.

Source code in import_export_extensions/api/mixins/import_mixins.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
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,
    )

Mixin which adds start export action.

Source code in import_export_extensions/api/mixins/export_mixins.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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,
            *args,
            **kwargs,
        ) -> 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) -> QuerySet:
        """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()  # type: ignore[misc]

    def get_export_detail_serializer_class(
        self,
    ) -> type[serializers.ExportJobSerializer]:
        """Get serializer which will be used show details of export job."""
        return self.export_detail_serializer_class

    def get_export_create_serializer_class(self) -> type:
        """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) -> Serializer:
        """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)  # type: ignore[misc]

    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,
        )

get_export_create_serializer_class()

Get serializer which will be used to start export job.

Source code in import_export_extensions/api/mixins/export_mixins.py 
111
112
113
114
115
def get_export_create_serializer_class(self) -> type:
    """Get serializer which will be used to start export job."""
    return serializers.get_create_export_job_serializer(
        self.resource_class,
    )

get_export_detail_serializer_class()

Get serializer which will be used show details of export job.

Source code in import_export_extensions/api/mixins/export_mixins.py 
105
106
107
108
109
def get_export_detail_serializer_class(
    self,
) -> type[serializers.ExportJobSerializer]:
    """Get serializer which will be used show details of export job."""
    return self.export_detail_serializer_class

get_export_resource_kwargs()

Provide extra arguments to resource class.

Source code in import_export_extensions/api/mixins/export_mixins.py 
117
118
119
def get_export_resource_kwargs(self) -> dict[str, typing.Any]:
    """Provide extra arguments to resource class."""
    return {}

get_queryset()

Return export model queryset on export action.

For better openapi support and consistency.

Source code in import_export_extensions/api/mixins/export_mixins.py 
 95
 96
 97
 98
 99
100
101
102
103
def get_queryset(self) -> QuerySet:
    """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()  # type: ignore[misc]

get_serializer(*args, **kwargs)

Provide resource kwargs to serializer class.

Source code in import_export_extensions/api/mixins/export_mixins.py 
121
122
123
124
125
126
127
128
def get_serializer(self, *args, **kwargs) -> Serializer:
    """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)  # type: ignore[misc]

start_export(request)

Validate request data and start ExportJob.

Source code in import_export_extensions/api/mixins/export_mixins.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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,
    )

Bases: Serializer

Base Serializer to start export job.

It used to set up base workflow of ExportJob creation via API.

Source code in import_export_extensions/api/serializers/export_job.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class CreateExportJob(serializers.Serializer):
    """Base Serializer to start export job.

    It used to set up base workflow of ExportJob creation via API.

    """

    resource_class: type[resources.CeleryModelResource]

    def __init__(
        self,
        *args,
        ordering: collections.abc.Sequence[str] | None = None,
        filter_kwargs: dict[str, typing.Any] | None = None,
        resource_kwargs: dict[str, typing.Any] | None = None,
        **kwargs,
    ) -> None:
        """Set ordering, filter kwargs and current user."""
        super().__init__(*args, **kwargs)
        self._ordering = ordering
        self._filter_kwargs = filter_kwargs
        self._resource_kwargs = resource_kwargs or {}
        self._request: request.Request = self.context.get("request")
        self._user = getattr(self._request, "user", None)

    def validate(self, attrs: dict[str, typing.Any]) -> dict[str, typing.Any]:
        """Check that ordering and filter kwargs are valid."""
        self.resource_class(
            ordering=self._ordering,
            filter_kwargs=self._filter_kwargs,
            created_by=self._user,
            **self._resource_kwargs,
        ).get_queryset()
        return attrs

    def create(
        self,
        validated_data: dict[str, typing.Any],
    ) -> models.ExportJob:
        """Create export job."""
        file_format_class = self.resource_class.get_supported_extensions_map()[
            validated_data["file_format"]
        ]
        return models.ExportJob.objects.create(
            resource_path=self.resource_class.class_path,
            file_format_path=f"{file_format_class.__module__}.{file_format_class.__name__}",
            resource_kwargs={
                "ordering": self._ordering,
                "filter_kwargs": self._filter_kwargs,
                **self._resource_kwargs,
            },
            created_by=self._user,
        )

    def update(self, instance, validated_data) -> None:
        """Empty method to pass linters checks."""

__init__(*args, ordering=None, filter_kwargs=None, resource_kwargs=None, **kwargs)

Set ordering, filter kwargs and current user.

Source code in import_export_extensions/api/serializers/export_job.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    *args,
    ordering: collections.abc.Sequence[str] | None = None,
    filter_kwargs: dict[str, typing.Any] | None = None,
    resource_kwargs: dict[str, typing.Any] | None = None,
    **kwargs,
) -> None:
    """Set ordering, filter kwargs and current user."""
    super().__init__(*args, **kwargs)
    self._ordering = ordering
    self._filter_kwargs = filter_kwargs
    self._resource_kwargs = resource_kwargs or {}
    self._request: request.Request = self.context.get("request")
    self._user = getattr(self._request, "user", None)

create(validated_data)

Create export job.

Source code in import_export_extensions/api/serializers/export_job.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def create(
    self,
    validated_data: dict[str, typing.Any],
) -> models.ExportJob:
    """Create export job."""
    file_format_class = self.resource_class.get_supported_extensions_map()[
        validated_data["file_format"]
    ]
    return models.ExportJob.objects.create(
        resource_path=self.resource_class.class_path,
        file_format_path=f"{file_format_class.__module__}.{file_format_class.__name__}",
        resource_kwargs={
            "ordering": self._ordering,
            "filter_kwargs": self._filter_kwargs,
            **self._resource_kwargs,
        },
        created_by=self._user,
    )

update(instance, validated_data)

Empty method to pass linters checks.

Source code in import_export_extensions/api/serializers/export_job.py 
98
99
def update(self, instance, validated_data) -> None:
    """Empty method to pass linters checks."""

validate(attrs)

Check that ordering and filter kwargs are valid.

Source code in import_export_extensions/api/serializers/export_job.py 
69
70
71
72
73
74
75
76
77
def validate(self, attrs: dict[str, typing.Any]) -> dict[str, typing.Any]:
    """Check that ordering and filter kwargs are valid."""
    self.resource_class(
        ordering=self._ordering,
        filter_kwargs=self._filter_kwargs,
        created_by=self._user,
        **self._resource_kwargs,
    ).get_queryset()
    return attrs

Bases: Serializer

Base Serializer to start import job.

It used to set up base workflow of ImportJob creation via API.

Source code in import_export_extensions/api/serializers/import_job.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class CreateImportJob(serializers.Serializer):
    """Base Serializer to start import job.

    It used to set up base workflow of ImportJob creation via API.

    """

    resource_class: type[resources.CeleryModelResource]

    file = serializers.FileField(required=True)
    force_import = serializers.BooleanField(default=False, required=False)
    skip_parse_step = serializers.BooleanField(default=False, required=False)

    def __init__(
        self,
        *args,
        resource_kwargs: dict[str, typing.Any] | None = None,
        **kwargs,
    ) -> None:
        """Set filter kwargs and current user."""
        super().__init__(*args, **kwargs)
        self._request: request.Request = self.context.get("request")
        self._resource_kwargs = resource_kwargs or {}
        self._user = getattr(self._request, "user", None)

    def create(
        self,
        validated_data: dict[str, typing.Any],
    ) -> models.ImportJob:
        """Create import job."""
        return models.ImportJob.objects.create(
            data_file=validated_data["file"],
            force_import=validated_data["force_import"],
            skip_parse_step=validated_data["skip_parse_step"],
            resource_path=self.resource_class.class_path,
            resource_kwargs=self._resource_kwargs,
            created_by=self._user,
        )

    def update(self, instance, validated_data) -> None:
        """Empty method to pass linters checks."""

__init__(*args, resource_kwargs=None, **kwargs)

Set filter kwargs and current user.

Source code in import_export_extensions/api/serializers/import_job.py 
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    *args,
    resource_kwargs: dict[str, typing.Any] | None = None,
    **kwargs,
) -> None:
    """Set filter kwargs and current user."""
    super().__init__(*args, **kwargs)
    self._request: request.Request = self.context.get("request")
    self._resource_kwargs = resource_kwargs or {}
    self._user = getattr(self._request, "user", None)

create(validated_data)

Create import job.

Source code in import_export_extensions/api/serializers/import_job.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def create(
    self,
    validated_data: dict[str, typing.Any],
) -> models.ImportJob:
    """Create import job."""
    return models.ImportJob.objects.create(
        data_file=validated_data["file"],
        force_import=validated_data["force_import"],
        skip_parse_step=validated_data["skip_parse_step"],
        resource_path=self.resource_class.class_path,
        resource_kwargs=self._resource_kwargs,
        created_by=self._user,
    )

update(instance, validated_data)

Empty method to pass linters checks.

Source code in import_export_extensions/api/serializers/import_job.py 
125
126
def update(self, instance, validated_data) -> None:
    """Empty method to pass linters checks."""

Bases: ModelSerializer

Serializer to show information about export job.

Source code in import_export_extensions/api/serializers/export_job.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class ExportJobSerializer(serializers.ModelSerializer):
    """Serializer to show information about export job."""

    progress = ExportProgressSerializer()

    class Meta:
        model = models.ExportJob
        fields = (
            "id",
            "export_status",
            "data_file",
            "progress",
            "export_started",
            "export_finished",
            "created",
            "modified",
        )

Bases: ModelSerializer

Serializer to show information about import job.

Source code in import_export_extensions/api/serializers/import_job.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class ImportJobSerializer(serializers.ModelSerializer):
    """Serializer to show information about import job."""

    progress = ImportProgressSerializer()

    import_params = details.ImportParamsSerializer(
        read_only=True,
        source="*",
    )
    totals = details.TotalsSerializer(
        read_only=True,
        source="*",
    )
    parse_error = serializers.CharField(
        source="error_message",
        read_only=True,
        allow_blank=True,
    )
    input_error = details.InputErrorSerializer(
        source="*",
        read_only=True,
    )
    skipped_errors = details.SkippedErrorsSerializer(
        source="*",
        read_only=True,
    )
    importing_data = details.ImportingDataSerializer(
        read_only=True,
        source="*",
    )
    input_errors_file = serializers.FileField(
        read_only=True,
        allow_null=True,
    )
    is_all_rows_shown = details.IsAllRowsShowField(
        source="*",
        read_only=True,
    )

    class Meta:
        model = models.ImportJob
        fields = (
            "id",
            "progress",
            "import_status",
            "import_params",
            "totals",
            "parse_error",
            "input_error",
            "skipped_errors",
            "is_all_rows_shown",
            "importing_data",
            "input_errors_file",
            "import_started",
            "import_finished",
            "force_import",
            "created",
            "modified",
        )

Bases: Serializer

Serializer to show progress of job.

Source code in import_export_extensions/api/serializers/progress.py 
11
12
13
14
class ProgressSerializer(serializers.Serializer):
    """Serializer to show progress of job."""

    info = ProgressInfoSerializer()

Bases: Serializer

Serializer to show progress info, like how much is done.

Source code in import_export_extensions/api/serializers/progress.py 
4
5
6
7
8
class ProgressInfoSerializer(serializers.Serializer):
    """Serializer to show progress info, like how much is done."""

    current = serializers.IntegerField()
    total = serializers.IntegerField()