Skip to content

Admin

ExportJobAdmin

Bases: BaseImportExportJobAdminMixin, ModelAdmin

Admin class for debugging ExportJob.

Source code in import_export_extensions/admin/model_admins/export_job_admin.py 
 16
 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
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
@admin.register(models.ExportJob)
class ExportJobAdmin(
    mixins.BaseImportExportJobAdminMixin,
    admin.ModelAdmin,
):
    """Admin class for debugging ExportJob."""

    form = forms.ExportJobAdminForm
    exclude = ("result",)
    list_display = (
        "id",
        "export_status",
        "_model",
        "created_by",
        "created",
        "export_started",
        "export_finished",
    )
    list_display_links = (
        "id",
        "export_status",
        "_model",
    )
    date_hierarchy = "created"
    export_job_model = models.ExportJob
    list_filter = ("export_status",)
    list_select_related = ("created_by",)
    actions = (
        "cancel_jobs",
    )
    readonly_fields = (
        "export_status",
        "traceback",
        "file_format_path",
        "created",
        "export_started",
        "export_finished",
        "error_message",
        "_model",
        "created_by",
        "resource_path",
        "data_file",
        "resource_kwargs",
    )

    def get_urls(self) -> list[URLPattern]:
        """Add url to get current export job progress in JSON representation.

        /admin/import_export_extensions/exportjob/<job_id>/progress/

        """
        urls = super().get_urls()
        export_urls = [
            re_path(
                route=r"^(?P<job_id>\d+)/progress/$",
                view=self.admin_site.admin_view(self.export_job_progress_view),
                name="export_job_progress",
            ),
        ]
        return export_urls + urls

    def export_job_progress_view(
        self,
        request: WSGIRequest,
        job_id: int,
        **kwargs,
    ) -> JsonResponse:
        """View to return `ExportJob` status as JSON.

        If current status is exporting, view also returns job state
        and percent of completed work.

        Return:
            Response: dictionary with status (optionally, state and percent).

        """
        try:
            job: models.ExportJob = self.export_job_model.objects.get(
                id=job_id,
            )
        except self.export_job_model.DoesNotExist as error:
            return JsonResponse(
                {"validation_error": error.args[0]},
                status=http.HTTPStatus.NOT_FOUND,
            )

        response_data = {"status": job.export_status.title()}

        if job.export_status != models.ExportJob.ExportStatus.EXPORTING:
            return JsonResponse(response_data)

        percent = 0
        total = 0
        current = 0
        job_progress = job.progress
        progress_info = job_progress["info"]

        if progress_info and progress_info["total"]:
            total = progress_info["total"]
            current = progress_info["current"]
            percent = int(100 / total * current)

        response_data.update(
            state=job_progress["state"],
            percent=percent,
            total=total,
            current=current,
        )
        return JsonResponse(response_data)

    def get_fieldsets(
        self,
        request: WSGIRequest,
        obj: models.ExportJob,
    ) -> list[tuple[str, dict[str, typing.Any]]]:
        """Get fieldsets depending on object status."""
        status = (
            _("Status"),
            {
                "fields": (
                    "export_status",
                    "_model",
                    "created_by",
                    "created",
                    "export_started",
                    "export_finished",
                ),
            },
        )
        progress = (
            _("Status"),
            {
                "fields": (
                    "export_status",
                    "export_progressbar",
                ),
            },
        )
        export_params = (
            _("Export params"),
            {
                "fields": (
                    "resource_path",
                    "resource_kwargs",
                    "file_format_path",
                ),
                "classes": ("collapse",),
            },
        )
        traceback_fields = (
            _("Traceback"),
            {
                "fields": (
                    "error_message",
                    "traceback",
                ),
            },
        )
        result = (
            _("Export results"),
            {
                "fields": ("data_file",),
            },
        )

        if obj.export_status == models.ExportJob.ExportStatus.CREATED:
            return [status, export_params]

        if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:
            return [status, result, export_params]

        if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:
            return [progress, export_params]

        return [status, traceback_fields, export_params]

    @admin.action(description="Cancel selected jobs")
    def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
        """Admin action for cancelling data export."""
        for job in queryset:
            try:
                job.cancel_export()
                self.message_user(
                    request,
                    _(f"Export of {job} canceled"),
                    messages.SUCCESS,
                )
            except ValueError as error:
                self.message_user(request, str(error), messages.ERROR)

cancel_jobs(request, queryset)

Admin action for cancelling data export.

Source code in import_export_extensions/admin/model_admins/export_job_admin.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
@admin.action(description="Cancel selected jobs")
def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
    """Admin action for cancelling data export."""
    for job in queryset:
        try:
            job.cancel_export()
            self.message_user(
                request,
                _(f"Export of {job} canceled"),
                messages.SUCCESS,
            )
        except ValueError as error:
            self.message_user(request, str(error), messages.ERROR)

export_job_progress_view(request, job_id, **kwargs)

View to return ExportJob status as JSON.

If current status is exporting, view also returns job state and percent of completed work.

Return

Response: dictionary with status (optionally, state and percent).

Source code in import_export_extensions/admin/model_admins/export_job_admin.py 
 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
def export_job_progress_view(
    self,
    request: WSGIRequest,
    job_id: int,
    **kwargs,
) -> JsonResponse:
    """View to return `ExportJob` status as JSON.

    If current status is exporting, view also returns job state
    and percent of completed work.

    Return:
        Response: dictionary with status (optionally, state and percent).

    """
    try:
        job: models.ExportJob = self.export_job_model.objects.get(
            id=job_id,
        )
    except self.export_job_model.DoesNotExist as error:
        return JsonResponse(
            {"validation_error": error.args[0]},
            status=http.HTTPStatus.NOT_FOUND,
        )

    response_data = {"status": job.export_status.title()}

    if job.export_status != models.ExportJob.ExportStatus.EXPORTING:
        return JsonResponse(response_data)

    percent = 0
    total = 0
    current = 0
    job_progress = job.progress
    progress_info = job_progress["info"]

    if progress_info and progress_info["total"]:
        total = progress_info["total"]
        current = progress_info["current"]
        percent = int(100 / total * current)

    response_data.update(
        state=job_progress["state"],
        percent=percent,
        total=total,
        current=current,
    )
    return JsonResponse(response_data)

get_fieldsets(request, obj)

Get fieldsets depending on object status.

Source code in import_export_extensions/admin/model_admins/export_job_admin.py 
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
def get_fieldsets(
    self,
    request: WSGIRequest,
    obj: models.ExportJob,
) -> list[tuple[str, dict[str, typing.Any]]]:
    """Get fieldsets depending on object status."""
    status = (
        _("Status"),
        {
            "fields": (
                "export_status",
                "_model",
                "created_by",
                "created",
                "export_started",
                "export_finished",
            ),
        },
    )
    progress = (
        _("Status"),
        {
            "fields": (
                "export_status",
                "export_progressbar",
            ),
        },
    )
    export_params = (
        _("Export params"),
        {
            "fields": (
                "resource_path",
                "resource_kwargs",
                "file_format_path",
            ),
            "classes": ("collapse",),
        },
    )
    traceback_fields = (
        _("Traceback"),
        {
            "fields": (
                "error_message",
                "traceback",
            ),
        },
    )
    result = (
        _("Export results"),
        {
            "fields": ("data_file",),
        },
    )

    if obj.export_status == models.ExportJob.ExportStatus.CREATED:
        return [status, export_params]

    if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:
        return [status, result, export_params]

    if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:
        return [progress, export_params]

    return [status, traceback_fields, export_params]

get_urls()

Add url to get current export job progress in JSON representation.

/admin/import_export_extensions/exportjob//progress/

Source code in import_export_extensions/admin/model_admins/export_job_admin.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_urls(self) -> list[URLPattern]:
    """Add url to get current export job progress in JSON representation.

    /admin/import_export_extensions/exportjob/<job_id>/progress/

    """
    urls = super().get_urls()
    export_urls = [
        re_path(
            route=r"^(?P<job_id>\d+)/progress/$",
            view=self.admin_site.admin_view(self.export_job_progress_view),
            name="export_job_progress",
        ),
    ]
    return export_urls + urls

ImportJobAdmin

Bases: BaseImportExportJobAdminMixin, ModelAdmin

Admin class for debugging ImportJob.

Source code in import_export_extensions/admin/model_admins/import_job_admin.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
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@admin.register(models.ImportJob)
class ImportJobAdmin(
    mixins.BaseImportExportJobAdminMixin,
    admin.ModelAdmin,
):
    """Admin class for debugging ImportJob."""

    form = forms.ImportJobAdminForm
    exclude = ("result",)
    list_display = (
        "id",
        "import_status",
        "_model",
        "created_by",
        "created",
        "parse_finished",
        "import_started",
        "import_finished",
    )
    list_display_links = (
        "id",
        "import_status",
        "_model",
    )
    date_hierarchy = "created"
    import_job_model = models.ImportJob
    list_filter = ("import_status",)
    list_select_related = ("created_by",)
    actions = (
        "cancel_jobs",
        "confirm_jobs",
    )
    readonly_fields = (
        "import_status",
        "_model",
        "created_by",
        "traceback",
        "_show_results",
        "_input_errors",
        "created",
        "parse_finished",
        "import_started",
        "import_finished",
        "resource_path",
        "input_errors_file",
        "data_file",
        "resource_kwargs",
    )

    def get_queryset(self, request: WSGIRequest) -> QuerySet:
        """Override `get_queryset`.

        Do not get `result` from db because it can be rather big and is not
        used in admin.

        """
        return super().get_queryset(request).defer("result")

    def get_urls(self) -> list[URLPattern]:
        """Add url to get current import job progress in JSON representation.

        /admin/import_export_extensions/importjob/<job_id>/progress/

        """
        urls = super().get_urls()
        import_urls = [
            re_path(
                r"^(?P<job_id>\d+)/progress/$",
                self.admin_site.admin_view(self.import_job_progress_view),
                name="import_job_progress",
            ),
        ]
        return import_urls + urls

    def import_job_progress_view(
        self,
        request: WSGIRequest,
        job_id: int,
        **kwargs,
    ) -> JsonResponse:
        """View to return ``ImportJob`` status as JSON.

        If current status is parsing/importing, view also returns job state
        and percent of completed work.

        Return:
            Response: dictionary with status (optionally, state and percent).

        """
        try:
            job: models.ImportJob = self.import_job_model.objects.get(
                id=job_id,
            )
        except self.import_job_model.DoesNotExist as error:
            return JsonResponse(
                {"validation_error": error.args[0]},
                status=http.HTTPStatus.NOT_FOUND,
            )

        response_data = {"status": job.import_status.title()}

        if job.import_status not in models.ImportJob.progress_statuses:
            return JsonResponse(response_data)

        percent = 0
        total = 0
        current = 0
        job_progress = job.progress
        progress_info = job_progress["info"]

        if progress_info and progress_info["total"]:
            total = progress_info["total"]
            current = progress_info["current"]
            percent = int(100 / total * current)

        response_data.update(
            state=job_progress["state"],
            percent=percent,
            total=total,
            current=current,
        )

        return JsonResponse(response_data)

    def _show_results(
        self,
        obj: models.ImportJob,
    ) -> str:
        """Show results totals.

        Example return value:
        New: 99
        Update: 1
        Delete: 0
        Skip: 0
        Error: 0

        """
        result_sections = []
        for key, value in obj.result.totals.items():
            status_template = f"{key.title()}: {value}"
            result_sections.append(status_template)

        return "\n".join(result_sections)

    _show_results.short_description = _("Parse/Import results")

    def _input_errors(self, job: models.ImportJob) -> SafeText:
        """Render html with input errors."""
        template = "admin/import_export_extensions/import_job_results.html"
        return render_to_string(
            template,
            {"result": job.result},
        )

    _input_errors.short_description = _("Import data")
    _input_errors.allow_tags = True

    def get_fieldsets(
        self,
        request: WSGIRequest,
        obj: models.ImportJob,
    ) -> list[tuple[str, dict[str, typing.Any]]]:
        """Get fieldsets depending on object status."""
        status = (
            _("Status"),
            {
                "fields": (
                    "import_status",
                    "_model",
                    "created_by",
                    "created",
                    "parse_finished",
                    "import_started",
                    "import_finished",
                ),
            },
        )
        progress = (
            _("Status"),
            {
                "fields": (
                    "import_status",
                    "import_progressbar",
                ),
            },
        )
        import_params = (
            _("Import params"),
            {
                "fields": (
                    "data_file",
                    "resource_path",
                    "resource_kwargs",
                ),
                "classes": ("collapse",),
            },
        )
        traceback_ = (
            _("Traceback"),
            {
                "fields": ("traceback",),
            },
        )
        result = (
            _("Result totals"),
            {
                "fields": ("_show_results",),
                "classes": ("collapse",),
            },
        )
        data = (
            _("Importing data"),
            {
                "fields": (
                    "input_errors_file",
                    "_input_errors",
                ),
                "classes": ("collapse",),
            },
        )

        if obj.import_status == models.ImportJob.ImportStatus.CREATED:
            return [status, import_params]

        if obj.import_status in models.ImportJob.success_statuses:
            return [status, result, data, import_params]

        if obj.import_status in models.ImportJob.progress_statuses:
            return [status, progress, import_params]

        return [status, traceback_, import_params]

    @admin.action(description="Cancel selected jobs")
    def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
        """Admin action for cancelling data import."""
        for job in queryset:
            try:
                job.cancel_import()
                self.message_user(
                    request,
                    _(f"Import of {job} canceled"),
                    messages.SUCCESS,
                )
            except ValueError as error:
                self.message_user(request, str(error), messages.ERROR)

    @admin.action(description="Confirm selected jobs")
    def confirm_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
        """Admin action for confirming data import."""
        for job in queryset:
            try:
                job.confirm_import()
                self.message_user(
                    request,
                    _(f"Import of {job} confirmed"),
                    messages.SUCCESS,
                )
            except ValueError as error:
                self.message_user(request, str(error), messages.ERROR)

cancel_jobs(request, queryset)

Admin action for cancelling data import.

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
@admin.action(description="Cancel selected jobs")
def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
    """Admin action for cancelling data import."""
    for job in queryset:
        try:
            job.cancel_import()
            self.message_user(
                request,
                _(f"Import of {job} canceled"),
                messages.SUCCESS,
            )
        except ValueError as error:
            self.message_user(request, str(error), messages.ERROR)

confirm_jobs(request, queryset)

Admin action for confirming data import.

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
@admin.action(description="Confirm selected jobs")
def confirm_jobs(self, request: WSGIRequest, queryset: QuerySet) -> None:
    """Admin action for confirming data import."""
    for job in queryset:
        try:
            job.confirm_import()
            self.message_user(
                request,
                _(f"Import of {job} confirmed"),
                messages.SUCCESS,
            )
        except ValueError as error:
            self.message_user(request, str(error), messages.ERROR)

get_fieldsets(request, obj)

Get fieldsets depending on object status.

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def get_fieldsets(
    self,
    request: WSGIRequest,
    obj: models.ImportJob,
) -> list[tuple[str, dict[str, typing.Any]]]:
    """Get fieldsets depending on object status."""
    status = (
        _("Status"),
        {
            "fields": (
                "import_status",
                "_model",
                "created_by",
                "created",
                "parse_finished",
                "import_started",
                "import_finished",
            ),
        },
    )
    progress = (
        _("Status"),
        {
            "fields": (
                "import_status",
                "import_progressbar",
            ),
        },
    )
    import_params = (
        _("Import params"),
        {
            "fields": (
                "data_file",
                "resource_path",
                "resource_kwargs",
            ),
            "classes": ("collapse",),
        },
    )
    traceback_ = (
        _("Traceback"),
        {
            "fields": ("traceback",),
        },
    )
    result = (
        _("Result totals"),
        {
            "fields": ("_show_results",),
            "classes": ("collapse",),
        },
    )
    data = (
        _("Importing data"),
        {
            "fields": (
                "input_errors_file",
                "_input_errors",
            ),
            "classes": ("collapse",),
        },
    )

    if obj.import_status == models.ImportJob.ImportStatus.CREATED:
        return [status, import_params]

    if obj.import_status in models.ImportJob.success_statuses:
        return [status, result, data, import_params]

    if obj.import_status in models.ImportJob.progress_statuses:
        return [status, progress, import_params]

    return [status, traceback_, import_params]

get_queryset(request)

Override get_queryset.

Do not get result from db because it can be rather big and is not used in admin.

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
67
68
69
70
71
72
73
74
def get_queryset(self, request: WSGIRequest) -> QuerySet:
    """Override `get_queryset`.

    Do not get `result` from db because it can be rather big and is not
    used in admin.

    """
    return super().get_queryset(request).defer("result")

get_urls()

Add url to get current import job progress in JSON representation.

/admin/import_export_extensions/importjob//progress/

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_urls(self) -> list[URLPattern]:
    """Add url to get current import job progress in JSON representation.

    /admin/import_export_extensions/importjob/<job_id>/progress/

    """
    urls = super().get_urls()
    import_urls = [
        re_path(
            r"^(?P<job_id>\d+)/progress/$",
            self.admin_site.admin_view(self.import_job_progress_view),
            name="import_job_progress",
        ),
    ]
    return import_urls + urls

import_job_progress_view(request, job_id, **kwargs)

View to return ImportJob status as JSON.

If current status is parsing/importing, view also returns job state and percent of completed work.

Return

Response: dictionary with status (optionally, state and percent).

Source code in import_export_extensions/admin/model_admins/import_job_admin.py 
 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
def import_job_progress_view(
    self,
    request: WSGIRequest,
    job_id: int,
    **kwargs,
) -> JsonResponse:
    """View to return ``ImportJob`` status as JSON.

    If current status is parsing/importing, view also returns job state
    and percent of completed work.

    Return:
        Response: dictionary with status (optionally, state and percent).

    """
    try:
        job: models.ImportJob = self.import_job_model.objects.get(
            id=job_id,
        )
    except self.import_job_model.DoesNotExist as error:
        return JsonResponse(
            {"validation_error": error.args[0]},
            status=http.HTTPStatus.NOT_FOUND,
        )

    response_data = {"status": job.import_status.title()}

    if job.import_status not in models.ImportJob.progress_statuses:
        return JsonResponse(response_data)

    percent = 0
    total = 0
    current = 0
    job_progress = job.progress
    progress_info = job_progress["info"]

    if progress_info and progress_info["total"]:
        total = progress_info["total"]
        current = progress_info["current"]
        percent = int(100 / total * current)

    response_data.update(
        state=job_progress["state"],
        percent=percent,
        total=total,
        current=current,
    )

    return JsonResponse(response_data)

BaseImportExportJobAdminMixin

Mixin provides common methods for ImportJob and ExportJob admins.

Source code in import_export_extensions/admin/model_admins/mixins.py 
 9
10
11
12
13
14
15
16
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
class BaseImportExportJobAdminMixin:
    """Mixin provides common methods for ImportJob and ExportJob admins."""

    def has_add_permission(
        self,
        request: WSGIRequest,
        *args,
        **kwargs,
    ) -> bool:
        """Import/Export Jobs should not be created using this interface."""
        return False

    def has_delete_permission(
        self,
        request: WSGIRequest,
        *args,
        **kwargs,
    ) -> bool:
        """Import/Export Jobs should not be deleted using this interface.

        Instead, admins must cancel jobs.

        """
        return False

    def _model(self, obj: BaseJob) -> str:
        """Add `model` field of import/export job."""
        try:
            resource_class = import_string(obj.resource_path)
            model = resource_class.Meta.model._meta.verbose_name_plural
        # In case resource has no Meta or model we need to catch AttributeError
        except (ImportError, AttributeError):  # pragma: no cover
            model = _("Unknown")
        return model

    _model.short_description = _("Model")

    def get_from_content_type(
        self,
        obj: BaseJob,
    ) -> ContentType | None:  # pragma: no cover
        """Shortcut to get object from content_type."""
        content_type = obj.resource_kwargs.get("content_type")
        obj_id = obj.resource_kwargs.get("object_id")

        if content_type and obj_id:
            content_type = ContentType.objects.get(id=content_type)
            return content_type.model_class().objects.filter(id=obj_id).first()
        return None

get_from_content_type(obj)

Shortcut to get object from content_type.

Source code in import_export_extensions/admin/model_admins/mixins.py 
46
47
48
49
50
51
52
53
54
55
56
57
def get_from_content_type(
    self,
    obj: BaseJob,
) -> ContentType | None:  # pragma: no cover
    """Shortcut to get object from content_type."""
    content_type = obj.resource_kwargs.get("content_type")
    obj_id = obj.resource_kwargs.get("object_id")

    if content_type and obj_id:
        content_type = ContentType.objects.get(id=content_type)
        return content_type.model_class().objects.filter(id=obj_id).first()
    return None

has_add_permission(request, *args, **kwargs)

Import/Export Jobs should not be created using this interface.

Source code in import_export_extensions/admin/model_admins/mixins.py 
12
13
14
15
16
17
18
19
def has_add_permission(
    self,
    request: WSGIRequest,
    *args,
    **kwargs,
) -> bool:
    """Import/Export Jobs should not be created using this interface."""
    return False

has_delete_permission(request, *args, **kwargs)

Import/Export Jobs should not be deleted using this interface.

Instead, admins must cancel jobs.

Source code in import_export_extensions/admin/model_admins/mixins.py 
21
22
23
24
25
26
27
28
29
30
31
32
def has_delete_permission(
    self,
    request: WSGIRequest,
    *args,
    **kwargs,
) -> bool:
    """Import/Export Jobs should not be deleted using this interface.

    Instead, admins must cancel jobs.

    """
    return False

ExportJobAdminForm

Bases: ModelForm

Admin form for ExportJob model.

Adds custom export_progressbar field that displays current export progress using AJAX requests to specified endpoint. Fields widget is defined in __init__ method.

Source code in import_export_extensions/admin/forms/export_job_admin_form.py 
 8
 9
10
11
12
13
14
15
16
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
class ExportJobAdminForm(forms.ModelForm):
    """Admin form for `ExportJob` model.

    Adds custom `export_progressbar` field that displays current export
    progress using AJAX requests to specified endpoint. Fields widget is
    defined in `__init__` method.

    """

    export_progressbar = forms.Field(
        label="Export progress",
        required=False,
    )

    def __init__(
        self,
        instance: models.ExportJob,
        *args,
        **kwargs,
    ) -> None:
        """Provide `export_progressbar` widget the `ExportJob` instance."""
        super().__init__(*args, instance=instance, **kwargs)
        url_name = "admin:export_job_progress"
        self.fields["export_progressbar"].widget = ProgressBarWidget(
            job=instance,
            url=reverse(url_name, args=(instance.id,)),
        )

    class Meta:
        fields = (
            "export_status",
            "resource_path",
            "file_format_path",
            "data_file",
            "resource_kwargs",
            "traceback",
            "error_message",
            "result",
            "export_task_id",
            "export_started",
            "export_finished",
            "created_by",
            "created",
            "modified",
        )

__init__(instance, *args, **kwargs)

Provide export_progressbar widget the ExportJob instance.

Source code in import_export_extensions/admin/forms/export_job_admin_form.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(
    self,
    instance: models.ExportJob,
    *args,
    **kwargs,
) -> None:
    """Provide `export_progressbar` widget the `ExportJob` instance."""
    super().__init__(*args, instance=instance, **kwargs)
    url_name = "admin:export_job_progress"
    self.fields["export_progressbar"].widget = ProgressBarWidget(
        job=instance,
        url=reverse(url_name, args=(instance.id,)),
    )

ImportJobAdminForm

Bases: ModelForm

Admin form for ImportJob model.

Adds custom import_progressbar field that displays current import progress using AJAX requests to specified endpoint. Fields widget is defined in __init__ method.

Source code in import_export_extensions/admin/forms/import_job_admin_form.py 
 8
 9
10
11
12
13
14
15
16
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
class ImportJobAdminForm(forms.ModelForm):
    """Admin form for ``ImportJob`` model.

    Adds custom `import_progressbar` field that displays current import
    progress using AJAX requests to specified endpoint. Fields widget is
    defined in `__init__` method.

    """

    import_progressbar = forms.Field(
        required=False,
    )

    def __init__(
        self,
        instance: models.ImportJob,
        *args,
        **kwargs,
    ) -> None:
        """Provide `import_progressbar` widget the ``ImportJob`` instance."""
        super().__init__(*args, instance=instance, **kwargs)
        url_name = "admin:import_job_progress"
        self.fields["import_progressbar"].label = (
            "Import progress" if
            instance.import_status == models.ImportJob.ImportStatus.IMPORTING
            else "Parsing progress"
        )
        self.fields["import_progressbar"].widget = ProgressBarWidget(
            job=instance,
            url=reverse(url_name, args=(instance.id,)),
        )

    class Meta:
        fields = (
            "import_status",
            "resource_path",
            "data_file",
            "resource_kwargs",
            "traceback",
            "error_message",
            "result",
            "parse_task_id",
            "import_task_id",
            "parse_finished",
            "import_started",
            "import_finished",
            "created_by",
            "created",
            "modified",
        )

__init__(instance, *args, **kwargs)

Provide import_progressbar widget the ImportJob instance.

Source code in import_export_extensions/admin/forms/import_job_admin_form.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    instance: models.ImportJob,
    *args,
    **kwargs,
) -> None:
    """Provide `import_progressbar` widget the ``ImportJob`` instance."""
    super().__init__(*args, instance=instance, **kwargs)
    url_name = "admin:import_job_progress"
    self.fields["import_progressbar"].label = (
        "Import progress" if
        instance.import_status == models.ImportJob.ImportStatus.IMPORTING
        else "Parsing progress"
    )
    self.fields["import_progressbar"].widget = ProgressBarWidget(
        job=instance,
        url=reverse(url_name, args=(instance.id,)),
    )

ForceImportForm

Bases: ImportForm

Import form with force_import option.

Source code in import_export_extensions/admin/forms/import_admin_form.py 
 6
 7
 8
 9
10
11
12
class ForceImportForm(base_forms.ImportForm):
    """Import form with `force_import` option."""

    force_import = forms.BooleanField(
        required=False,
        initial=False,
    )

ModelInfo dataclass

Contain base info about imported model.

Source code in import_export_extensions/admin/mixins/types.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@dataclasses.dataclass
class ModelInfo:
    """Contain base info about imported model."""

    meta: Options

    @property
    def name(self) -> str:
        """Get name of model."""
        return self.meta.model_name

    @property
    def app_label(self) -> str:
        """App label of model."""
        return self.meta.app_label

    @property
    def app_model_name(self) -> str:
        """Return url name."""
        return f"{self.app_label}_{self.name}"

app_label property

App label of model.

app_model_name property

Return url name.

name property

Get name of model.

CeleryExportAdminMixin

Bases: BaseExportMixin, BaseCeleryImportExportAdminMixin

Admin mixin for celery export.

Admin export work-flow is

GET celery_export_action() - display form with format type input

POST celery_export_action() - create ExportJob and starts data export This view redirects to next view:

GET celery_export_job_status_view() - display ExportJob status (with progress bar). When data exporting is done, redirect to next view:

GET celery_export_job_results_view() - display export results. If no errors - success message and link to the file with exported data. If errors - traceback and error message.

Source code in import_export_extensions/admin/mixins/export_mixin.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
 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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
class CeleryExportAdminMixin(
    import_export_mixins.BaseExportMixin,
    base_mixin.BaseCeleryImportExportAdminMixin,
):
    """Admin mixin for celery export.

    Admin export work-flow is:
        GET `celery_export_action()` - display form with format type input

        POST `celery_export_action()` - create ExportJob and starts data export
            This view redirects to next view:

        GET `celery_export_job_status_view()` - display ExportJob status (with
            progress bar). When data exporting is done, redirect to next view:

        GET `celery_export_job_results_view()` - display export results. If no
            errors - success message and link to the file with exported data.
            If errors - traceback and error message.

    """

    # export data encoding
    to_encoding = "utf-8"

    export_form_class: type[import_export_forms.ExportForm] = (
        import_export_forms.SelectableFieldsExportForm
    )

    # template used to display ExportForm
    celery_export_template_name = "admin/import_export/export.html"

    export_status_template_name = (
        "admin/import_export_extensions/celery_export_status.html"
    )

    export_results_template_name = (
        "admin/import_export_extensions/celery_export_results.html"
    )

    import_export_change_list_template = (
        "admin/import_export/change_list_export.html"
    )

    # Statuses that should be displayed on 'results' page
    export_results_statuses = models.ExportJob.export_finished_statuses

    # Copy methods of mixin from original package to reuse it here
    has_export_permission = (
        import_export_admin.ExportMixin.has_export_permission
    )
    get_export_form_class = import_export_admin.ExportMixin.get_export_form_class  # noqa
    get_export_resource_fields_from_form = import_export_admin.ExportMixin.get_export_resource_fields_from_form  # noqa
    is_skip_export_form_enabled = import_export_admin.ExportMixin.is_skip_export_form_enabled  # noqa

    def get_export_context_data(self, **kwargs) -> dict[str, typing.Any]:
        """Get context data for export."""
        return self.get_context_data(**kwargs)

    def get_urls(self) -> list[URLPattern]:
        """Return list of urls.

        /<model/celery-export/:
            ExportForm ('export_action' method)
        /<model>/celery-export/<ID>/:
            status of ExportJob and progress bar ('export_job_status_view')
        /<model>/celery-export/<ID>/results/:
            table with export results (errors)

        """
        urls = super().get_urls()
        export_urls = [
            re_path(
                r"^celery-export/$",
                self.admin_site.admin_view(self.celery_export_action),
                name=f"{self.model_info.app_model_name}_export",
            ),
            re_path(
                r"^celery-export/(?P<job_id>\d+)/$",
                self.admin_site.admin_view(self.export_job_status_view),
                name=(
                    f"{self.model_info.app_model_name}"
                    f"_export_job_status"
                ),
            ),
            re_path(
                r"^celery-export/(?P<job_id>\d+)/results/$",
                self.admin_site.admin_view(
                    self.export_job_results_view,
                ),
                name=(
                    f"{self.model_info.app_model_name}"
                    f"_export_job_results"
                ),
            ),
        ]
        return export_urls + urls

    def celery_export_action(
        self,
        request: WSGIRequest,
        *args,
        **kwargs,
    ) -> HttpResponse | TemplateResponse:
        """Show and handle export.

        GET: show export form with format_type input
        POST: create ExportJob instance and redirect to it's status

        """
        if not self.has_export_permission(request):
            raise PermissionDenied

        formats = self.get_export_formats()
        resource_kwargs = self.get_export_resource_kwargs(
            *args,
            **kwargs,
            request=request,
        )
        if self.is_skip_export_form_enabled():
            job = self.create_export_job(
                request=request,
                resource_class=self.get_export_resource_classes(request)[0],
                resource_kwargs=resource_kwargs,
                file_format=formats[0],
            )
            return self._redirect_to_export_status_page(
                request=request,
                job=job,
            )

        form_type = self.get_export_form_class()
        form = form_type(
            formats=formats,
            resources=self.get_export_resource_classes(request),
            data=request.POST or None,
        )

        if request.method == "POST" and form.is_valid():
            file_format = formats[int(form.cleaned_data["format"])]
            # Get the selected export fields from the form
            resource_kwargs["export_fields"] = (
                self.get_export_resource_fields_from_form(form)
            )
            # create ExportJob and redirect to page with it's status
            job = self.create_export_job(
                request=request,
                resource_class=self.choose_export_resource_class(
                    form,
                    request,
                ),
                resource_kwargs=resource_kwargs,
                file_format=file_format,
            )
            return self._redirect_to_export_status_page(
                request=request,
                job=job,
            )

        # GET: display Export Form
        context = self.get_export_context_data()
        context.update(self.admin_site.each_context(request))

        context["title"] = _("Export")
        context["form"] = form
        context["opts"] = self.model_info.meta
        request.current_app = self.admin_site.name

        return TemplateResponse(
            request=request,
            template=[self.celery_export_template_name],
            context=context,
        )

    def export_job_status_view(
        self,
        request: WSGIRequest,
        job_id: int,
        **kwargs,
    ) -> HttpResponse:
        """View to track export job status.

        Displays current export job status and progress (using JS + another
        view).

        If job result is ready - redirects to another page to see results.

        """
        if not self.has_export_permission(request):
            raise PermissionDenied

        job = self.get_export_job(request=request, job_id=job_id)
        if job.export_status in self.export_results_statuses:
            return self._redirect_to_export_results_page(
                request=request,
                job=job,
            )

        context = self.get_export_context_data()
        job_url = reverse("admin:export_job_progress", args=(job.id,))

        context["title"] = _("Export status")
        context["opts"] = self.model_info.meta
        context["export_job"] = job
        context["export_job_url"] = job_url
        request.current_app = self.admin_site.name
        return TemplateResponse(
            request=request,
            template=[self.export_status_template_name],
            context=context,
        )

    def export_job_results_view(
        self,
        request: WSGIRequest,
        job_id: int,
        *args,
        **kwargs,
    ) -> HttpResponse:
        """Display export results.

        GET-request:
            * show message
            * if no errors - show file link
            * if errors - show traceback and error

        """
        if not self.has_export_permission(request):
            raise PermissionDenied

        job = self.get_export_job(request=request, job_id=job_id)
        if job.export_status not in self.export_results_statuses:
            return self._redirect_to_export_status_page(
                request=request,
                job=job,
            )

        context = self.get_export_context_data()

        # GET request, show export results
        context["title"] = _("Export results")
        context["opts"] = self.model._meta
        context["export_job"] = job
        context["result"] = job.export_status

        return TemplateResponse(
            request=request,
            template=[self.export_results_template_name],
            context=context,
        )

    def create_export_job(
        self,
        request: WSGIRequest,
        resource_class: types.ResourceType,
        resource_kwargs: dict[str, typing.Any],
        file_format: types.FormatType,
    ) -> models.ExportJob:
        """Create and return instance of export job with chosen format."""
        job = models.ExportJob.objects.create(
            resource_path=resource_class.class_path,
            resource_kwargs=resource_kwargs,
            created_by=request.user,
            file_format_path=(
                f"{file_format.__module__}.{file_format.__name__}"
            ),
        )
        return job

    def get_export_job(
        self,
        request: WSGIRequest,
        job_id: int,
    ) -> models.ExportJob:
        """Get ExportJob instance.

        Raises
            Http404

        """
        return get_object_or_404(models.ExportJob, id=job_id)

    def get_resource_kwargs(
        self,
        request: WSGIRequest,
        *args,
        **kwargs,
    ) -> dict[str, typing.Any]:
        """Return filter kwargs for resource queryset."""
        resource_kwargs = super().get_resource_kwargs(request, *args, **kwargs)
        resource_kwargs["admin_filters"] = {
            "pk__in": list(
                self.get_changelist_instance(request)
                .get_queryset(request)
                .values_list("pk", flat=True),
            ),
        }
        return resource_kwargs

    def _redirect_to_export_status_page(
        self,
        request: WSGIRequest,
        job: models.ExportJob,
    ) -> HttpResponse:
        """Shortcut for redirecting to job's status page."""
        url_name = (
            f"admin:{self.model_info.app_model_name}_export_job_status"
        )
        url = reverse(url_name, kwargs={"job_id": job.id})
        query = request.GET.urlencode()
        url = f"{url}?{query}" if query else url
        return HttpResponseRedirect(redirect_to=url)

    def _redirect_to_export_results_page(
        self,
        request: WSGIRequest,
        job: models.ExportJob,
    ) -> HttpResponse:
        """Shortcut for redirecting to job's results page."""
        url_name = (
            f"admin:{self.model_info.app_model_name}_export_job_results"
        )
        url = reverse(url_name, kwargs={"job_id": job.id})
        query = request.GET.urlencode()
        url = f"{url}?{query}" if query else url
        return HttpResponseRedirect(redirect_to=url)

    def changelist_view(
        self,
        request: WSGIRequest,
        context: dict[str, typing.Any] | None = None,
    ) -> HttpResponse:
        """Add the check for permission to changelist template context."""
        context = context or {}
        context["has_export_permission"] = self.has_export_permission(request)
        return super().changelist_view(request, context)

celery_export_action(request, *args, **kwargs)

Show and handle export.

GET: show export form with format_type input POST: create ExportJob instance and redirect to it's status

Source code in import_export_extensions/admin/mixins/export_mixin.py 
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
def celery_export_action(
    self,
    request: WSGIRequest,
    *args,
    **kwargs,
) -> HttpResponse | TemplateResponse:
    """Show and handle export.

    GET: show export form with format_type input
    POST: create ExportJob instance and redirect to it's status

    """
    if not self.has_export_permission(request):
        raise PermissionDenied

    formats = self.get_export_formats()
    resource_kwargs = self.get_export_resource_kwargs(
        *args,
        **kwargs,
        request=request,
    )
    if self.is_skip_export_form_enabled():
        job = self.create_export_job(
            request=request,
            resource_class=self.get_export_resource_classes(request)[0],
            resource_kwargs=resource_kwargs,
            file_format=formats[0],
        )
        return self._redirect_to_export_status_page(
            request=request,
            job=job,
        )

    form_type = self.get_export_form_class()
    form = form_type(
        formats=formats,
        resources=self.get_export_resource_classes(request),
        data=request.POST or None,
    )

    if request.method == "POST" and form.is_valid():
        file_format = formats[int(form.cleaned_data["format"])]
        # Get the selected export fields from the form
        resource_kwargs["export_fields"] = (
            self.get_export_resource_fields_from_form(form)
        )
        # create ExportJob and redirect to page with it's status
        job = self.create_export_job(
            request=request,
            resource_class=self.choose_export_resource_class(
                form,
                request,
            ),
            resource_kwargs=resource_kwargs,
            file_format=file_format,
        )
        return self._redirect_to_export_status_page(
            request=request,
            job=job,
        )

    # GET: display Export Form
    context = self.get_export_context_data()
    context.update(self.admin_site.each_context(request))

    context["title"] = _("Export")
    context["form"] = form
    context["opts"] = self.model_info.meta
    request.current_app = self.admin_site.name

    return TemplateResponse(
        request=request,
        template=[self.celery_export_template_name],
        context=context,
    )

changelist_view(request, context=None)

Add the check for permission to changelist template context.

Source code in import_export_extensions/admin/mixins/export_mixin.py 
348
349
350
351
352
353
354
355
356
def changelist_view(
    self,
    request: WSGIRequest,
    context: dict[str, typing.Any] | None = None,
) -> HttpResponse:
    """Add the check for permission to changelist template context."""
    context = context or {}
    context["has_export_permission"] = self.has_export_permission(request)
    return super().changelist_view(request, context)

create_export_job(request, resource_class, resource_kwargs, file_format)

Create and return instance of export job with chosen format.

Source code in import_export_extensions/admin/mixins/export_mixin.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def create_export_job(
    self,
    request: WSGIRequest,
    resource_class: types.ResourceType,
    resource_kwargs: dict[str, typing.Any],
    file_format: types.FormatType,
) -> models.ExportJob:
    """Create and return instance of export job with chosen format."""
    job = models.ExportJob.objects.create(
        resource_path=resource_class.class_path,
        resource_kwargs=resource_kwargs,
        created_by=request.user,
        file_format_path=(
            f"{file_format.__module__}.{file_format.__name__}"
        ),
    )
    return job

export_job_results_view(request, job_id, *args, **kwargs)

Display export results.

GET-request
  • show message
  • if no errors - show file link
  • if errors - show traceback and error
Source code in import_export_extensions/admin/mixins/export_mixin.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def export_job_results_view(
    self,
    request: WSGIRequest,
    job_id: int,
    *args,
    **kwargs,
) -> HttpResponse:
    """Display export results.

    GET-request:
        * show message
        * if no errors - show file link
        * if errors - show traceback and error

    """
    if not self.has_export_permission(request):
        raise PermissionDenied

    job = self.get_export_job(request=request, job_id=job_id)
    if job.export_status not in self.export_results_statuses:
        return self._redirect_to_export_status_page(
            request=request,
            job=job,
        )

    context = self.get_export_context_data()

    # GET request, show export results
    context["title"] = _("Export results")
    context["opts"] = self.model._meta
    context["export_job"] = job
    context["result"] = job.export_status

    return TemplateResponse(
        request=request,
        template=[self.export_results_template_name],
        context=context,
    )

export_job_status_view(request, job_id, **kwargs)

View to track export job status.

Displays current export job status and progress (using JS + another view).

If job result is ready - redirects to another page to see results.

Source code in import_export_extensions/admin/mixins/export_mixin.py 
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
231
def export_job_status_view(
    self,
    request: WSGIRequest,
    job_id: int,
    **kwargs,
) -> HttpResponse:
    """View to track export job status.

    Displays current export job status and progress (using JS + another
    view).

    If job result is ready - redirects to another page to see results.

    """
    if not self.has_export_permission(request):
        raise PermissionDenied

    job = self.get_export_job(request=request, job_id=job_id)
    if job.export_status in self.export_results_statuses:
        return self._redirect_to_export_results_page(
            request=request,
            job=job,
        )

    context = self.get_export_context_data()
    job_url = reverse("admin:export_job_progress", args=(job.id,))

    context["title"] = _("Export status")
    context["opts"] = self.model_info.meta
    context["export_job"] = job
    context["export_job_url"] = job_url
    request.current_app = self.admin_site.name
    return TemplateResponse(
        request=request,
        template=[self.export_status_template_name],
        context=context,
    )

get_export_context_data(**kwargs)

Get context data for export.

Source code in import_export_extensions/admin/mixins/export_mixin.py 
76
77
78
def get_export_context_data(self, **kwargs) -> dict[str, typing.Any]:
    """Get context data for export."""
    return self.get_context_data(**kwargs)

get_export_job(request, job_id)

Get ExportJob instance.

Raises Http404

Source code in import_export_extensions/admin/mixins/export_mixin.py 
290
291
292
293
294
295
296
297
298
299
300
301
def get_export_job(
    self,
    request: WSGIRequest,
    job_id: int,
) -> models.ExportJob:
    """Get ExportJob instance.

    Raises
        Http404

    """
    return get_object_or_404(models.ExportJob, id=job_id)

get_resource_kwargs(request, *args, **kwargs)

Return filter kwargs for resource queryset.

Source code in import_export_extensions/admin/mixins/export_mixin.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def get_resource_kwargs(
    self,
    request: WSGIRequest,
    *args,
    **kwargs,
) -> dict[str, typing.Any]:
    """Return filter kwargs for resource queryset."""
    resource_kwargs = super().get_resource_kwargs(request, *args, **kwargs)
    resource_kwargs["admin_filters"] = {
        "pk__in": list(
            self.get_changelist_instance(request)
            .get_queryset(request)
            .values_list("pk", flat=True),
        ),
    }
    return resource_kwargs

get_urls()

Return list of urls.

/<model/celery-export/: ExportForm ('export_action' method) //celery-export//: status of ExportJob and progress bar ('export_job_status_view') //celery-export//results/: table with export results (errors)

Source code in import_export_extensions/admin/mixins/export_mixin.py 
 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
def get_urls(self) -> list[URLPattern]:
    """Return list of urls.

    /<model/celery-export/:
        ExportForm ('export_action' method)
    /<model>/celery-export/<ID>/:
        status of ExportJob and progress bar ('export_job_status_view')
    /<model>/celery-export/<ID>/results/:
        table with export results (errors)

    """
    urls = super().get_urls()
    export_urls = [
        re_path(
            r"^celery-export/$",
            self.admin_site.admin_view(self.celery_export_action),
            name=f"{self.model_info.app_model_name}_export",
        ),
        re_path(
            r"^celery-export/(?P<job_id>\d+)/$",
            self.admin_site.admin_view(self.export_job_status_view),
            name=(
                f"{self.model_info.app_model_name}"
                f"_export_job_status"
            ),
        ),
        re_path(
            r"^celery-export/(?P<job_id>\d+)/results/$",
            self.admin_site.admin_view(
                self.export_job_results_view,
            ),
            name=(
                f"{self.model_info.app_model_name}"
                f"_export_job_results"
            ),
        ),
    ]
    return export_urls + urls

CeleryImportExportMixin

Bases: CeleryImportAdminMixin, CeleryExportAdminMixin

Import and export mixin.

Source code in import_export_extensions/admin/mixins/import_export_mixin.py 
 5
 6
 7
 8
 9
10
11
12
13
14
class CeleryImportExportMixin(
    CeleryImportAdminMixin,
    CeleryExportAdminMixin,
):
    """Import and export mixin."""

    # template for change_list view
    import_export_change_list_template = (
        "admin/import_export/change_list_import_export.html"
    )

CeleryImportAdminMixin

Bases: BaseImportMixin, BaseCeleryImportExportAdminMixin

Admin mixin for celery import.

Admin import work-flow is:

GET celery_import_action() - display form with import file input

POST celery_import_action() - save file and create ImportJob. This view redirects to next view:

GET celery_import_job_status_view() - display ImportJob status (with progress bar and critical errors occurred). When data parsing is done, redirect to next view:

GET celery_import_job_results_view() - display rows that will be imported and data parse errors. If no errors - next step. If errors - display same form as in import_action()

POST celery_import_job_results_view() - start data importing and redirect back to GET celery_import_job_status_view() with progress bar and import totals.

Source code in import_export_extensions/admin/mixins/import_mixin.py 
 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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
class CeleryImportAdminMixin(
    import_export_mixins.BaseImportMixin,
    base_mixin.BaseCeleryImportExportAdminMixin,
):
    """Admin mixin for celery import.

    Admin import work-flow is:

    GET ``celery_import_action()`` - display form with import file input

    POST ``celery_import_action()`` - save file and create ImportJob.
        This view redirects to next view:

    GET ``celery_import_job_status_view()`` - display ImportJob status (with
        progress bar and critical errors occurred). When data parsing is
        done, redirect to next view:

    GET ``celery_import_job_results_view()`` - display rows that will be
        imported and data parse errors. If no errors - next step.
        If errors - display same form as in ``import_action()``

    POST ``celery_import_job_results_view()`` - start data importing and
        redirect back to GET ``celery_import_job_status_view()``
        with progress bar and import totals.

    """

    # Import data encoding
    from_encoding = "utf-8"

    import_form_class: type[ImportForm] = ForceImportForm
    confirm_form_class: type[ConfirmImportForm] = ConfirmImportForm

    # Statuses that should be displayed on 'results' page
    results_statuses = models.ImportJob.results_statuses

    # Template used to display ImportForm
    celery_import_template = "admin/import_export/import.html"

    # Template used to display status of import jobs
    import_status_template = (
        "admin/import_export_extensions/celery_import_status.html"
    )

    # template used to display results of import jobs
    import_result_template_name = (
        "admin/import_export_extensions/celery_import_results.html"
    )

    import_export_change_list_template = (
        "admin/import_export/change_list_import.html"
    )

    skip_admin_log = None
    # Copy methods of mixin from original package to reuse it here
    generate_log_entries = import_export_admin.ImportMixin.generate_log_entries
    get_skip_admin_log = import_export_admin.ImportMixin.get_skip_admin_log
    has_import_permission = (
        import_export_admin.ImportMixin.has_import_permission
    )
    _log_actions = import_export_admin.ImportMixin._log_actions
    _create_log_entries = import_export_admin.ImportMixin._create_log_entries
    _create_log_entry = import_export_admin.ImportMixin._create_log_entry

    # Copied form related methods
    create_import_form = import_export_admin.ImportMixin.create_import_form
    get_import_form_class = import_export_admin.ImportMixin.get_import_form_class  # noqa
    get_import_form_kwargs = import_export_admin.ImportMixin.get_import_form_kwargs  # noqa
    get_import_form_initial = import_export_admin.ImportMixin.get_import_form_initial  # noqa
    create_confirm_form = import_export_admin.ImportMixin.create_confirm_form
    get_confirm_form_class = import_export_admin.ImportMixin.get_confirm_form_class  # noqa
    get_confirm_form_kwargs = import_export_admin.ImportMixin.get_confirm_form_kwargs  # noqa
    get_confirm_form_initial = import_export_admin.ImportMixin.get_confirm_form_initial  # noqa

    def get_import_context_data(self, **kwargs) -> dict[str, typing.Any]:
        """Get context data for import."""
        return self.get_context_data(**kwargs)

    def get_urls(self) -> list[URLPattern]:
        """Return list of urls.

        * /<model>/<celery-import>/:
            ImportForm ('celery_import_action' method)
        * /<model>/<celery-import>/<ID>/:
            status of ImportJob and progress bar
            ('celery_import_job_status_view')
        * /<model>/<celery-import>/<ID>/results/:
            table with import results (errors) and import confirmation
            ('celery_import_job_results_view')

        """
        urls = super().get_urls()
        import_urls = [
            re_path(
                r"^celery-import/$",
                self.admin_site.admin_view(self.celery_import_action),
                name=f"{self.model_info.app_model_name}_import",
            ),
            re_path(
                r"^celery-import/(?P<job_id>\d+)/$",
                self.admin_site.admin_view(self.celery_import_job_status_view),
                name=(
                    f"{self.model_info.app_model_name}"
                    f"_import_job_status"
                ),
            ),
            re_path(
                r"^celery-import/(?P<job_id>\d+)/results/$",
                self.admin_site.admin_view(
                    self.celery_import_job_results_view,
                ),
                name=(
                    f"{self.model_info.app_model_name}"
                    f"_import_job_results"
                ),
            ),
        ]
        return import_urls + urls

    def celery_import_action(
        self,
        request: WSGIRequest,
        *args,
        **kwargs,
    ) -> HttpResponseRedirect | TemplateResponse:
        """Show and handle ImportForm.

        GET:
            show import form with data_file input form
        POST:
            create ImportJob instance and redirect to it's status

        """
        if not self.has_import_permission(request):
            raise PermissionDenied

        context = self.get_import_context_data()
        resource_classes = self.get_import_resource_classes(request)

        form = self.create_import_form(request)
        resource_kwargs = self.get_import_resource_kwargs(request)

        if request.method == "POST" and form.is_valid():
            # create ImportJob and redirect to page with it's status
            resource_class = self.choose_import_resource_class(form, request)
            job = self.create_import_job(
                request=request,
                resource=resource_class(**resource_kwargs),
                form=form,
            )
            return self._redirect_to_import_status_page(
                request=request,
                job=job,
            )

        # GET: display Import Form
        context.update(self.admin_site.each_context(request))

        context["title"] = _("Import")
        context["form"] = form
        context["opts"] = self.model_info.meta
        context["media"] = self.media + form.media
        context["fields_list"] = self._get_fields_list_for_resources(
            resource_classes=resource_classes,
            resource_kwargs=resource_kwargs,
        )

        request.current_app = self.admin_site.name
        return TemplateResponse(
            request,
            [self.celery_import_template],
            context,
        )

    def celery_import_job_status_view(
        self,
        request: WSGIRequest,
        job_id: int,
        **kwargs,
    ) -> HttpResponse:
        """View to track import job status.

        Displays current import job status and progress (using JS + another
        view).

        If job result is ready - redirects to another page to see results.

        Also generates admin log entries if the job has `IMPORTED` status.

        """
        if not self.has_import_permission(request):
            raise PermissionDenied

        job = self.get_import_job(request, job_id)
        if job.import_status in self.results_statuses:
            if job.import_status == models.ImportJob.ImportStatus.IMPORTED:
                self.generate_log_entries(job.result, request)
            return self._redirect_to_results_page(
                request=request,
                job=job,
            )

        context = self.get_import_context_data()
        job_url = reverse("admin:import_job_progress", args=(job.id,))
        context.update(
            {
                "title": _("Import status"),
                "opts": self.model_info.meta,
                "import_job": job,
                "import_job_url": job_url,
            },
        )
        request.current_app = self.admin_site.name
        return TemplateResponse(
            request=request,
            template=[self.import_status_template],
            context=context,
        )

    def celery_import_job_results_view(
        self,
        request: WSGIRequest,
        job_id: int,
        *args,
        **kwargs,
    ) -> HttpResponse:
        """Display table with import results and import confirm form.

        GET-request:
            * show row results
            * if data valid - show import confirmation form
            * if data invalid - show ImportForm for uploading other file

        POST-request:
            * start data importing if data is correct

        """
        if not self.has_import_permission(request):
            raise PermissionDenied

        job = self.get_import_job(request=request, job_id=job_id)
        if job.import_status not in self.results_statuses:
            return self._redirect_to_import_status_page(
                request=request,
                job=job,
            )

        context = self.get_import_context_data()

        if request.method == "GET":
            # GET request, show parse results
            result = job.result
            context["import_job"] = job
            context["result"] = result
            context["title"] = _("Import results")

            if job.import_status == models.ImportJob.ImportStatus.PARSED:
                context["confirm_form"] = self.create_confirm_form(request)
            else:
                # display import form
                resource_classes = self.get_import_resource_classes(request)
                resource_kwargs = self.get_import_resource_kwargs(request)

                context["import_form"] = self.create_import_form(request)
                context["fields_list"] = self._get_fields_list_for_resources(
                    resource_classes=resource_classes,
                    resource_kwargs=resource_kwargs,
                )

            context.update(self.admin_site.each_context(request))
            context["opts"] = self.model_info.meta
            request.current_app = self.admin_site.name
            return TemplateResponse(
                request,
                [self.import_result_template_name],
                context,
            )

        # POST request
        if job.import_status == models.ImportJob.ImportStatus.PARSED:
            # start celery task for data importing
            job.confirm_import()
            return self._redirect_to_import_status_page(
                request=request,
                job=job,
            )

        return HttpResponseForbidden(
            "Data invalid, before importing data "
            "needs to be successfully parsed. "
            f"Current status: {job.import_status}",
        )

    def create_import_job(
        self,
        request: WSGIRequest,
        form: Form,
        resource: types.ResourceObj,
    ) -> models.ImportJob:
        """Create and return instance of import job."""
        return models.ImportJob.objects.create(
            resource_path=resource.class_path,
            data_file=form.cleaned_data["import_file"],
            resource_kwargs=resource.resource_init_kwargs,
            created_by=request.user,
            skip_parse_step=getattr(
                settings,
                "IMPORT_EXPORT_SKIP_ADMIN_CONFIRM",
                False,
            ),
            force_import=form.cleaned_data.get("force_import", False),
        )

    def get_import_job(
        self,
        request: WSGIRequest,
        job_id: int,
    ) -> models.ImportJob:
        """Get ImportJob instance.

        Raises
            Http404

        """
        return get_object_or_404(klass=models.ImportJob, id=job_id)

    def _redirect_to_import_status_page(
        self,
        request: WSGIRequest,
        job: models.ImportJob,
    ) -> HttpResponseRedirect:
        """Shortcut for redirecting to job's status page."""
        url_name = (
            f"admin:{self.model_info.app_model_name}_import_job_status"
        )
        url = reverse(url_name, kwargs={"job_id": job.id})
        query = request.GET.urlencode()
        url = f"{url}?{query}" if query else url
        return HttpResponseRedirect(redirect_to=url)

    def _redirect_to_results_page(
        self,
        request: WSGIRequest,
        job: models.ImportJob,
    ) -> HttpResponseRedirect:
        """Shortcut for redirecting to job's results page."""
        url_name = (
            f"admin:{self.model_info.app_model_name}_import_job_results"
        )
        url = reverse(url_name, kwargs={"job_id": job.id})
        query = request.GET.urlencode()
        url = f"{url}?{query}" if query else url
        if job.import_status != models.ImportJob.ImportStatus.PARSED:
            return HttpResponseRedirect(redirect_to=url)

        return HttpResponseRedirect(redirect_to=url)

    def _get_fields_list_for_resources(
        self,
        resource_classes: list[type[import_export_resources.ModelResource]],
        resource_kwargs,
    ) -> list[tuple[str, list[str]]]:
        """Get fields list for resource classes."""
        resources = [
            resource_class(**resource_kwargs)
            for resource_class in resource_classes
        ]
        return [
            (
                resource.get_display_name(),
                [
                    field.column_name
                    for field in resource.get_user_visible_fields()
                ],
            )
            for resource in resources
        ]

    def changelist_view(
        self,
        request: WSGIRequest,
        context: dict[str, typing.Any] | None = None,
    ) -> HttpResponse:
        """Add the check for permission to changelist template context."""
        context = context or {}
        context["has_import_permission"] = self.has_import_permission(request)
        return super().changelist_view(request, context)

celery_import_action(request, *args, **kwargs)

Show and handle ImportForm.

GET

show import form with data_file input form

POST: create ImportJob instance and redirect to it's status

Source code in import_export_extensions/admin/mixins/import_mixin.py 
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
def celery_import_action(
    self,
    request: WSGIRequest,
    *args,
    **kwargs,
) -> HttpResponseRedirect | TemplateResponse:
    """Show and handle ImportForm.

    GET:
        show import form with data_file input form
    POST:
        create ImportJob instance and redirect to it's status

    """
    if not self.has_import_permission(request):
        raise PermissionDenied

    context = self.get_import_context_data()
    resource_classes = self.get_import_resource_classes(request)

    form = self.create_import_form(request)
    resource_kwargs = self.get_import_resource_kwargs(request)

    if request.method == "POST" and form.is_valid():
        # create ImportJob and redirect to page with it's status
        resource_class = self.choose_import_resource_class(form, request)
        job = self.create_import_job(
            request=request,
            resource=resource_class(**resource_kwargs),
            form=form,
        )
        return self._redirect_to_import_status_page(
            request=request,
            job=job,
        )

    # GET: display Import Form
    context.update(self.admin_site.each_context(request))

    context["title"] = _("Import")
    context["form"] = form
    context["opts"] = self.model_info.meta
    context["media"] = self.media + form.media
    context["fields_list"] = self._get_fields_list_for_resources(
        resource_classes=resource_classes,
        resource_kwargs=resource_kwargs,
    )

    request.current_app = self.admin_site.name
    return TemplateResponse(
        request,
        [self.celery_import_template],
        context,
    )

celery_import_job_results_view(request, job_id, *args, **kwargs)

Display table with import results and import confirm form.

GET-request
  • show row results
  • if data valid - show import confirmation form
  • if data invalid - show ImportForm for uploading other file
POST-request
  • start data importing if data is correct
Source code in import_export_extensions/admin/mixins/import_mixin.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def celery_import_job_results_view(
    self,
    request: WSGIRequest,
    job_id: int,
    *args,
    **kwargs,
) -> HttpResponse:
    """Display table with import results and import confirm form.

    GET-request:
        * show row results
        * if data valid - show import confirmation form
        * if data invalid - show ImportForm for uploading other file

    POST-request:
        * start data importing if data is correct

    """
    if not self.has_import_permission(request):
        raise PermissionDenied

    job = self.get_import_job(request=request, job_id=job_id)
    if job.import_status not in self.results_statuses:
        return self._redirect_to_import_status_page(
            request=request,
            job=job,
        )

    context = self.get_import_context_data()

    if request.method == "GET":
        # GET request, show parse results
        result = job.result
        context["import_job"] = job
        context["result"] = result
        context["title"] = _("Import results")

        if job.import_status == models.ImportJob.ImportStatus.PARSED:
            context["confirm_form"] = self.create_confirm_form(request)
        else:
            # display import form
            resource_classes = self.get_import_resource_classes(request)
            resource_kwargs = self.get_import_resource_kwargs(request)

            context["import_form"] = self.create_import_form(request)
            context["fields_list"] = self._get_fields_list_for_resources(
                resource_classes=resource_classes,
                resource_kwargs=resource_kwargs,
            )

        context.update(self.admin_site.each_context(request))
        context["opts"] = self.model_info.meta
        request.current_app = self.admin_site.name
        return TemplateResponse(
            request,
            [self.import_result_template_name],
            context,
        )

    # POST request
    if job.import_status == models.ImportJob.ImportStatus.PARSED:
        # start celery task for data importing
        job.confirm_import()
        return self._redirect_to_import_status_page(
            request=request,
            job=job,
        )

    return HttpResponseForbidden(
        "Data invalid, before importing data "
        "needs to be successfully parsed. "
        f"Current status: {job.import_status}",
    )

celery_import_job_status_view(request, job_id, **kwargs)

View to track import job status.

Displays current import job status and progress (using JS + another view).

If job result is ready - redirects to another page to see results.

Also generates admin log entries if the job has IMPORTED status.

Source code in import_export_extensions/admin/mixins/import_mixin.py 
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def celery_import_job_status_view(
    self,
    request: WSGIRequest,
    job_id: int,
    **kwargs,
) -> HttpResponse:
    """View to track import job status.

    Displays current import job status and progress (using JS + another
    view).

    If job result is ready - redirects to another page to see results.

    Also generates admin log entries if the job has `IMPORTED` status.

    """
    if not self.has_import_permission(request):
        raise PermissionDenied

    job = self.get_import_job(request, job_id)
    if job.import_status in self.results_statuses:
        if job.import_status == models.ImportJob.ImportStatus.IMPORTED:
            self.generate_log_entries(job.result, request)
        return self._redirect_to_results_page(
            request=request,
            job=job,
        )

    context = self.get_import_context_data()
    job_url = reverse("admin:import_job_progress", args=(job.id,))
    context.update(
        {
            "title": _("Import status"),
            "opts": self.model_info.meta,
            "import_job": job,
            "import_job_url": job_url,
        },
    )
    request.current_app = self.admin_site.name
    return TemplateResponse(
        request=request,
        template=[self.import_status_template],
        context=context,
    )

changelist_view(request, context=None)

Add the check for permission to changelist template context.

Source code in import_export_extensions/admin/mixins/import_mixin.py 
405
406
407
408
409
410
411
412
413
def changelist_view(
    self,
    request: WSGIRequest,
    context: dict[str, typing.Any] | None = None,
) -> HttpResponse:
    """Add the check for permission to changelist template context."""
    context = context or {}
    context["has_import_permission"] = self.has_import_permission(request)
    return super().changelist_view(request, context)

create_import_job(request, form, resource)

Create and return instance of import job.

Source code in import_export_extensions/admin/mixins/import_mixin.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def create_import_job(
    self,
    request: WSGIRequest,
    form: Form,
    resource: types.ResourceObj,
) -> models.ImportJob:
    """Create and return instance of import job."""
    return models.ImportJob.objects.create(
        resource_path=resource.class_path,
        data_file=form.cleaned_data["import_file"],
        resource_kwargs=resource.resource_init_kwargs,
        created_by=request.user,
        skip_parse_step=getattr(
            settings,
            "IMPORT_EXPORT_SKIP_ADMIN_CONFIRM",
            False,
        ),
        force_import=form.cleaned_data.get("force_import", False),
    )

get_import_context_data(**kwargs)

Get context data for import.

Source code in import_export_extensions/admin/mixins/import_mixin.py 
101
102
103
def get_import_context_data(self, **kwargs) -> dict[str, typing.Any]:
    """Get context data for import."""
    return self.get_context_data(**kwargs)

get_import_job(request, job_id)

Get ImportJob instance.

Raises Http404

Source code in import_export_extensions/admin/mixins/import_mixin.py 
340
341
342
343
344
345
346
347
348
349
350
351
def get_import_job(
    self,
    request: WSGIRequest,
    job_id: int,
) -> models.ImportJob:
    """Get ImportJob instance.

    Raises
        Http404

    """
    return get_object_or_404(klass=models.ImportJob, id=job_id)

get_urls()

Return list of urls.

  • ///: ImportForm ('celery_import_action' method)
  • ////: status of ImportJob and progress bar ('celery_import_job_status_view')
  • ////results/: table with import results (errors) and import confirmation ('celery_import_job_results_view')
Source code in import_export_extensions/admin/mixins/import_mixin.py 
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
def get_urls(self) -> list[URLPattern]:
    """Return list of urls.

    * /<model>/<celery-import>/:
        ImportForm ('celery_import_action' method)
    * /<model>/<celery-import>/<ID>/:
        status of ImportJob and progress bar
        ('celery_import_job_status_view')
    * /<model>/<celery-import>/<ID>/results/:
        table with import results (errors) and import confirmation
        ('celery_import_job_results_view')

    """
    urls = super().get_urls()
    import_urls = [
        re_path(
            r"^celery-import/$",
            self.admin_site.admin_view(self.celery_import_action),
            name=f"{self.model_info.app_model_name}_import",
        ),
        re_path(
            r"^celery-import/(?P<job_id>\d+)/$",
            self.admin_site.admin_view(self.celery_import_job_status_view),
            name=(
                f"{self.model_info.app_model_name}"
                f"_import_job_status"
            ),
        ),
        re_path(
            r"^celery-import/(?P<job_id>\d+)/results/$",
            self.admin_site.admin_view(
                self.celery_import_job_results_view,
            ),
            name=(
                f"{self.model_info.app_model_name}"
                f"_import_job_results"
            ),
        ),
    ]
    return import_urls + urls

ProgressBarWidget

Bases: Widget

Widget for progress bar field.

Value for progress_bar element is changed using JS code.

Source code in import_export_extensions/admin/widgets.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
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
class ProgressBarWidget(forms.Widget):
    """Widget for progress bar field.

    Value for `progress_bar` element is changed using JS code.

    """

    template_name = "admin/import_export_extensions/progress_bar.html"

    def __init__(self, *args, **kwargs) -> None:
        """Get ``ImportJob`` or ``ExportJob`` instance from kwargs.

        ``ImportJob`` or ``ExportJob`` instance is used
        to render hidden element in `render` method.

        """
        self.job = kwargs.pop("job")
        self.url = kwargs.pop("url")
        super().__init__(*args, **kwargs)

    def render(self, *args, **kwargs) -> str:
        """Render HTML5 `progress` element.

        Additionally, method provides hidden `import_job_url` and
        `export_job_url` value that is used in `js/admin/progress_bar.js`
        to send GET requests.

        """
        return render_to_string(self.template_name, {"job_url": self.url})

    class Media:
        """Class with custom assets for widget."""

        css = {
            "all": ("import_export_extensions/css/widgets/progress_bar.css",),
        }
        js = (
            "admin/js/jquery.init.js",
            "import_export_extensions/js/widgets/progress_bar.js",
        )

Media

Class with custom assets for widget.

Source code in import_export_extensions/admin/widgets.py 
36
37
38
39
40
41
42
43
44
45
class Media:
    """Class with custom assets for widget."""

    css = {
        "all": ("import_export_extensions/css/widgets/progress_bar.css",),
    }
    js = (
        "admin/js/jquery.init.js",
        "import_export_extensions/js/widgets/progress_bar.js",
    )

__init__(*args, **kwargs)

Get ImportJob or ExportJob instance from kwargs.

ImportJob or ExportJob instance is used to render hidden element in render method.

Source code in import_export_extensions/admin/widgets.py 
15
16
17
18
19
20
21
22
23
24
def __init__(self, *args, **kwargs) -> None:
    """Get ``ImportJob`` or ``ExportJob`` instance from kwargs.

    ``ImportJob`` or ``ExportJob`` instance is used
    to render hidden element in `render` method.

    """
    self.job = kwargs.pop("job")
    self.url = kwargs.pop("url")
    super().__init__(*args, **kwargs)

render(*args, **kwargs)

Render HTML5 progress element.

Additionally, method provides hidden import_job_url and export_job_url value that is used in js/admin/progress_bar.js to send GET requests.

Source code in import_export_extensions/admin/widgets.py 
26
27
28
29
30
31
32
33
34
def render(self, *args, **kwargs) -> str:
    """Render HTML5 `progress` element.

    Additionally, method provides hidden `import_job_url` and
    `export_job_url` value that is used in `js/admin/progress_bar.js`
    to send GET requests.

    """
    return render_to_string(self.template_name, {"job_url": self.url})