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 
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
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 
132
133
134
135
136
137
138
139
140
141
142
143
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 
 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
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 
105
106
107
108
109
110
111
112
113
114
115
116
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 
146
147
148
149
150
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 
119
120
121
122
123
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 
 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
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
    )
    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()
                )
            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 
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
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()
            )
        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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 
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
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
    )
    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()
                )
            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 
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
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()
            )
        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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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 
153
154
155
156
157
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 
126
127
128
129
130
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 
 17
 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:
        """Set up `start_import_action` as action of viewset."""
        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,
        )

__init_subclass__()

Set up start_import_action as action of viewset.

Source code in import_export_extensions/api/mixins/import_mixins.py 
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
def __init_subclass__(cls) -> None:
    """Set up `start_import_action` as action of viewset."""
    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)

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 
 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class ExportStartActionMixin:
    """Mixin which adds start export action."""

    resource_class: type[resources.CeleryModelResource]
    is_export_action: bool = False
    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_permission_classes: collections.abc.Sequence[
        type[drf_permissions.BasePermission]
    ] = ()
    export_throttle_classes: collections.abc.Sequence[
        type[drf_throttling.BaseThrottle]
    ] = ()
    throttle_scope: str | None = None
    export_throttle_scope: str = "export"
    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:
        """Set up `start_export_action` as action of viewset."""
        super().__init_subclass__()
        # Skip if it is has no resource_class specified
        if not hasattr(cls, "resource_class"):
            return

        def start_export_action(
            self: "ExportStartActionMixin",
            request: drf_request.Request,
            *args,
            **kwargs,
        ) -> drf_response.Response:
            return self.start_export(request)

        setattr(cls, cls.export_action, start_export_action)
        drf_decorators.action(
            methods=["POST"],
            detail=False,
            is_export_action=True,
            export_action_name=cls.export_action_name,
            export_action_url=cls.export_action_url,
            resource_class=cls.resource_class,
        )(getattr(cls, cls.export_action))

        for action_name in dir(cls):
            action_attr = getattr(cls, action_name)
            action_kwargs = getattr(action_attr, "kwargs", {})
            if action_kwargs.get("is_export_action"):
                cls.set_export_action(
                    action_name=action_name,
                    action_attr=action_attr,
                    action_kwargs=action_kwargs,
                )

    @classmethod
    def set_export_action(
        cls,
        action_name: str,
        action_attr: typing.Any,
        action_kwargs: dict[str, typing.Any],
    ) -> None:
        """Set up export action with proper attrs."""
        resource_class = action_kwargs["resource_class"]
        action_view = cls(**action_kwargs)
        filter_backends = [
            module_loading.import_string(
                settings.DRF_EXPORT_DJANGO_FILTERS_BACKEND,
            ),
        ]
        if action_view.export_ordering_fields:
            filter_backends.append(
                module_loading.import_string(
                    settings.DRF_EXPORT_ORDERING_BACKEND,
                ),
            )
        default_throttle_classes = (
            tuple(
                map(
                    module_loading.import_string,
                    settings.DRF_EXPORT_THROTTLE_CLASSES,
                ),
            )
            or drf_settings.api_settings.DEFAULT_THROTTLE_CLASSES
        )
        default_permission_classes = (
            tuple(
                map(
                    module_loading.import_string,
                    settings.DRF_EXPORT_PERMISSION_CLASSES,
                ),
            )
            or drf_settings.api_settings.DEFAULT_PERMISSION_CLASSES
        )
        action_attr.url_name = action_kwargs.get(
            "export_action_name",
            action_attr.url_name,
        )
        action_attr.url_path = action_kwargs.get(
            "export_action_url",
            action_attr.url_path,
        )
        action_kwargs.update(
            queryset=resource_class.get_model_queryset(),
            serializer_class=action_view.get_export_create_serializer_class(),
            filter_backends=filter_backends,
            filterset_class=getattr(resource_class, "filterset_class", None),
            ordering=action_view.export_ordering,
            ordering_fields=action_view.export_ordering_fields,
            permission_classes=(
                action_view.export_permission_classes
                or default_permission_classes
            ),
            throttle_classes=(
                action_view.export_throttle_classes or default_throttle_classes
            ),
            throttle_scope=action_view.export_throttle_scope,
        )
        cls.set_up_export_action_api_specs(
            action_name=action_name,
            action_view=action_view,
        )

    @classmethod
    def set_up_export_action_api_specs(
        cls,
        action_name: str,
        action_view: "ExportStartActionMixin",
    ) -> None:
        """Correct specs of drf-spectacular if it is installed."""
        with contextlib.suppress(ImportError):
            from drf_spectacular import utils

            utils.extend_schema_view(
                **{
                    action_name: utils.extend_schema(
                        description=action_view.export_open_api_description,
                        filters=True,
                        responses={
                            drf_status.HTTP_201_CREATED: action_view.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.is_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[drf_serializers.Serializer]:
        """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) -> drf_serializers.Serializer:
        """Provide resource kwargs to serializer class."""
        if self.is_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: drf_request.Request,
    ) -> drf_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 drf_response.Response(
            data=self.get_export_detail_serializer_class()(
                instance=export_job,
            ).data,
            status=drf_status.HTTP_201_CREATED,
        )

__init_subclass__()

Set up start_export_action as action of viewset.

Source code in import_export_extensions/api/mixins/export_mixins.py 
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
def __init_subclass__(cls) -> None:
    """Set up `start_export_action` as action of viewset."""
    super().__init_subclass__()
    # Skip if it is has no resource_class specified
    if not hasattr(cls, "resource_class"):
        return

    def start_export_action(
        self: "ExportStartActionMixin",
        request: drf_request.Request,
        *args,
        **kwargs,
    ) -> drf_response.Response:
        return self.start_export(request)

    setattr(cls, cls.export_action, start_export_action)
    drf_decorators.action(
        methods=["POST"],
        detail=False,
        is_export_action=True,
        export_action_name=cls.export_action_name,
        export_action_url=cls.export_action_url,
        resource_class=cls.resource_class,
    )(getattr(cls, cls.export_action))

    for action_name in dir(cls):
        action_attr = getattr(cls, action_name)
        action_kwargs = getattr(action_attr, "kwargs", {})
        if action_kwargs.get("is_export_action"):
            cls.set_export_action(
                action_name=action_name,
                action_attr=action_attr,
                action_kwargs=action_kwargs,
            )

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 
189
190
191
192
193
194
195
def get_export_create_serializer_class(
    self,
) -> type[drf_serializers.Serializer]:
    """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 
183
184
185
186
187
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 
197
198
199
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 
173
174
175
176
177
178
179
180
181
def get_queryset(self) -> QuerySet:
    """Return export model queryset on export action.

    For better openapi support and consistency.

    """
    if self.is_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 
201
202
203
204
205
206
207
208
def get_serializer(self, *args, **kwargs) -> drf_serializers.Serializer:
    """Provide resource kwargs to serializer class."""
    if self.is_export_action:
        kwargs.setdefault(
            "resource_kwargs",
            self.get_export_resource_kwargs(),
        )
    return super().get_serializer(*args, **kwargs)  # type: ignore[misc]

set_export_action(action_name, action_attr, action_kwargs) classmethod

Set up export action with proper attrs.

Source code in import_export_extensions/api/mixins/export_mixins.py 
 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
148
149
@classmethod
def set_export_action(
    cls,
    action_name: str,
    action_attr: typing.Any,
    action_kwargs: dict[str, typing.Any],
) -> None:
    """Set up export action with proper attrs."""
    resource_class = action_kwargs["resource_class"]
    action_view = cls(**action_kwargs)
    filter_backends = [
        module_loading.import_string(
            settings.DRF_EXPORT_DJANGO_FILTERS_BACKEND,
        ),
    ]
    if action_view.export_ordering_fields:
        filter_backends.append(
            module_loading.import_string(
                settings.DRF_EXPORT_ORDERING_BACKEND,
            ),
        )
    default_throttle_classes = (
        tuple(
            map(
                module_loading.import_string,
                settings.DRF_EXPORT_THROTTLE_CLASSES,
            ),
        )
        or drf_settings.api_settings.DEFAULT_THROTTLE_CLASSES
    )
    default_permission_classes = (
        tuple(
            map(
                module_loading.import_string,
                settings.DRF_EXPORT_PERMISSION_CLASSES,
            ),
        )
        or drf_settings.api_settings.DEFAULT_PERMISSION_CLASSES
    )
    action_attr.url_name = action_kwargs.get(
        "export_action_name",
        action_attr.url_name,
    )
    action_attr.url_path = action_kwargs.get(
        "export_action_url",
        action_attr.url_path,
    )
    action_kwargs.update(
        queryset=resource_class.get_model_queryset(),
        serializer_class=action_view.get_export_create_serializer_class(),
        filter_backends=filter_backends,
        filterset_class=getattr(resource_class, "filterset_class", None),
        ordering=action_view.export_ordering,
        ordering_fields=action_view.export_ordering_fields,
        permission_classes=(
            action_view.export_permission_classes
            or default_permission_classes
        ),
        throttle_classes=(
            action_view.export_throttle_classes or default_throttle_classes
        ),
        throttle_scope=action_view.export_throttle_scope,
    )
    cls.set_up_export_action_api_specs(
        action_name=action_name,
        action_view=action_view,
    )

set_up_export_action_api_specs(action_name, action_view) classmethod

Correct specs of drf-spectacular if it is installed.

Source code in import_export_extensions/api/mixins/export_mixins.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@classmethod
def set_up_export_action_api_specs(
    cls,
    action_name: str,
    action_view: "ExportStartActionMixin",
) -> None:
    """Correct specs of drf-spectacular if it is installed."""
    with contextlib.suppress(ImportError):
        from drf_spectacular import utils

        utils.extend_schema_view(
            **{
                action_name: utils.extend_schema(
                    description=action_view.export_open_api_description,
                    filters=True,
                    responses={
                        drf_status.HTTP_201_CREATED: action_view.get_export_detail_serializer_class(),  # noqa: E501
                    },
                ),
            },
        )(cls)

start_export(request)

Validate request data and start ExportJob.

Source code in import_export_extensions/api/mixins/export_mixins.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def start_export(
    self,
    request: drf_request.Request,
) -> drf_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 drf_response.Response(
        data=self.get_export_detail_serializer_class()(
            instance=export_job,
        ).data,
        status=drf_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 
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
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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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 
97
98
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 
68
69
70
71
72
73
74
75
76
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 
 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
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 
 98
 99
100
101
102
103
104
105
106
107
108
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
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 
124
125
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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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 
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
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()