6144128305

Committed 11 Sep 2023 08:50AM UTC coverage: 78.082%. Remained the same

Build # 6144128305

Build Type

push

github

Committed by

NikAzanov

Commit Message

Remove refs from Readme since pypi doesn't allow

Run Details

1083 of 1387 relevant lines covered (78.08%)

9.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.0

/import_export_extensions/admin/model_admins/import_job_admin.py

import typing

from django.contrib import admin, messages
from django.core.handlers.wsgi import WSGIRequest
from django.db.models import QuerySet
from django.http import JsonResponse
from django.template.loader import render_to_string
from django.urls import re_path
from django.utils.translation import gettext_lazy as _

from ... import models
from .. import forms
from . import mixins


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",
    )
    import_job_model = models.ImportJob
    list_filter = ("import_status",)
    list_select_related = ("created_by",)
    actions = (
        "cancel_jobs",
        "confirm_jobs",
    )

    def get_queryset(self, request: WSGIRequest):
        """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):
        """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(dict(validation_error=error.args[0]))

        response_data = dict(status=job.import_status.title())

        if job.import_status in models.ImportJob.progress_statuses:
            percent = 0
            total = 0
            current = 0
            info = job.progress["info"]

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

            response_data.update(
                dict(
                    state=job.progress["state"],
                    percent=percent,
                    total=total,
                    current=current,
                ),
            )

        return JsonResponse(response_data)

    def get_readonly_fields(
        self,
        request: WSGIRequest,
        obj: typing.Optional[models.ImportJob] = None,
    ) -> list[str]:
        """Return readonly fields.

        Some fields are editable for new ImportJob.

        """
        readonly_fields = [
            "import_status",
            "_model",
            "created_by",
            "traceback_str",
            "_show_results",
            "_input_errors",
            "created",
            "parse_finished",
            "import_started",
            "import_finished",
        ]
        if obj:
            readonly_fields.extend(
                [
                    "resource_path",
                    "data_file",
                    "resource_kwargs",
                ],
            )

        return readonly_fields

    def _show_results(
        self,
        obj: typing.Optional[models.ImportJob] = None,
    ) -> str:
        """Show results totals.

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

        """
        if not obj:
            return ""

        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):
        """Render html with input errors."""
        template = "admin/import_export_extensions/import_job_results.html"
        return render_to_string(
            template,
            dict(result=job.result),
        )

    _input_errors.short_description = "Import data"
    _input_errors.allow_tags = True

    def get_fieldsets(
        self,
        request: WSGIRequest,
        obj: typing.Optional[models.ImportJob] = None,
    ):
        """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",),
                "classes": ("collapse",),
            },
        )

        if not obj:
            return [status, import_params]

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

        if obj.import_status in models.ImportJob.results_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):
        """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):
        """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)


admin.site.register(models.ImportJob, ImportJobAdmin)

1	import typing	12✔
2
3	from django.contrib import admin, messages	12✔
4	from django.core.handlers.wsgi import WSGIRequest	12✔
5	from django.db.models import QuerySet	12✔
6	from django.http import JsonResponse	12✔
7	from django.template.loader import render_to_string	12✔
8	from django.urls import re_path	12✔
9	from django.utils.translation import gettext_lazy as _	12✔
10
11	from ... import models	12✔
12	from .. import forms	12✔
13	from . import mixins	12✔
14
15
16	class ImportJobAdmin(	12✔
17	mixins.BaseImportExportJobAdminMixin,
18	admin.ModelAdmin,
19	):
20	"""Admin class for debugging ImportJob."""
21
22	form = forms.ImportJobAdminForm	12✔
23	exclude = ("result",)	12✔
24	list_display = (	12✔
25	"id",
26	"import_status",
27	"_model",
28	"created_by",
29	"created",
30	"parse_finished",
31	"import_started",
32	"import_finished",
33	)
34	list_display_links = (	12✔
35	"id",
36	"import_status",
37	"_model",
38	)
39	import_job_model = models.ImportJob	12✔
40	list_filter = ("import_status",)	12✔
41	list_select_related = ("created_by",)	12✔
42	actions = (	12✔
43	"cancel_jobs",
44	"confirm_jobs",
45	)
46
47	def get_queryset(self, request: WSGIRequest):	12✔
48	"""Override `get_queryset`.
49
50	Do not get `result` from db because it can be rather big and is not
51	used in admin.
52
53	"""
54	return super().get_queryset(request).defer("result")	×
55
56	def get_urls(self):	12✔
57	"""Add url to get current import job progress in JSON representation.
58
59	/admin/import_export_extensions/importjob/<job_id>/progress/
60
61	"""
62	urls = super().get_urls()	12✔
63	import_urls = [	12✔
64	re_path(
65	r"^(?P<job_id>\d+)/progress/$",
66	self.admin_site.admin_view(self.import_job_progress_view),
67	name="import_job_progress",
68	),
69	]
70	return import_urls + urls	12✔
71
72	def import_job_progress_view(	12✔
73	self,
74	request: WSGIRequest,
75	job_id: int,
76	**kwargs,
77	) -> JsonResponse:
78	"""View to return ``ImportJob`` status as JSON.
79
80	If current status is parsing/importing, view also returns job state
81	and percent of completed work.
82
83	Return:
84	Response: dictionary with status (optionally, state and percent).
85
86	"""
87	try:	×
88	job: models.ImportJob = self.import_job_model.objects.get(	×
89	id=job_id,
90	)
91	except self.import_job_model.DoesNotExist as error:	×
92	return JsonResponse(dict(validation_error=error.args[0]))	×
93
94	response_data = dict(status=job.import_status.title())	×
95
96	if job.import_status in models.ImportJob.progress_statuses:	×
97	percent = 0	×
98	total = 0	×
99	current = 0	×
100	info = job.progress["info"]	×
101
102	if info and info["total"]:	×
103	percent = int(100 / info["total"] * info["current"])	×
104	total = info["total"]	×
105	current = info["current"]	×
106
107	response_data.update(	×
108	dict(
109	state=job.progress["state"],
110	percent=percent,
111	total=total,
112	current=current,
113	),
114	)
115
116	return JsonResponse(response_data)	×
117
118	def get_readonly_fields(	12✔
119	self,
120	request: WSGIRequest,
121	obj: typing.Optional[models.ImportJob] = None,
122	) -> list[str]:
123	"""Return readonly fields.
124
125	Some fields are editable for new ImportJob.
126
127	"""
128	readonly_fields = [	×
129	"import_status",
130	"_model",
131	"created_by",
132	"traceback_str",
133	"_show_results",
134	"_input_errors",
135	"created",
136	"parse_finished",
137	"import_started",
138	"import_finished",
139	]
140	if obj:	×
141	readonly_fields.extend(	×
142	[
143	"resource_path",
144	"data_file",
145	"resource_kwargs",
146	],
147	)
148
149	return readonly_fields	×
150
151	def _show_results(	12✔
152	self,
153	obj: typing.Optional[models.ImportJob] = None,
154	) -> str:
155	"""Show results totals.
156
157	Example return value:
158	New: 99
159	Update: 1
160	Delete: 0
161	Skip: 0
162	Error: 0
163
164	"""
165	if not obj:	×
166	return ""	×
167
168	result_sections = []	×
169	for key, value in obj.result.totals.items():	×
170	status_template = f"{key.title()}: {value}"	×
171	result_sections.append(status_template)	×
172
173	return "\n".join(result_sections)	×
174
175	_show_results.short_description = _("Parse/Import results")	12✔
176
177	def _input_errors(self, job: models.ImportJob):	12✔
178	"""Render html with input errors."""
179	template = "admin/import_export_extensions/import_job_results.html"	×
180	return render_to_string(	×
181	template,
182	dict(result=job.result),
183	)
184
185	_input_errors.short_description = "Import data"	12✔
186	_input_errors.allow_tags = True	12✔
187
188	def get_fieldsets(	12✔
189	self,
190	request: WSGIRequest,
191	obj: typing.Optional[models.ImportJob] = None,
192	):
193	"""Get fieldsets depending on object status."""
194	status = (	×
195	_("Status"),
196	{
197	"fields": (
198	"import_status",
199	"_model",
200	"created_by",
201	"created",
202	"parse_finished",
203	"import_started",
204	"import_finished",
205	),
206	},
207	)
208	progress = (	×
209	_("Status"),
210	{
211	"fields": (
212	"import_status",
213	"import_progressbar",
214	),
215	},
216	)
217	import_params = (	×
218	_("Import params"),
219	{
220	"fields": (
221	"data_file",
222	"resource_path",
223	"resource_kwargs",
224	),
225	"classes": ("collapse",),
226	},
227	)
228	traceback_ = (	×
229	_("Traceback"),
230	{
231	"fields": ("traceback",),
232	},
233	)
234	result = (	×
235	_("Result totals"),
236	{
237	"fields": ("_show_results",),
238	"classes": ("collapse",),
239	},
240	)
241	data = (	×
242	_("Importing data"),
243	{
244	"fields": ("_input_errors",),
245	"classes": ("collapse",),
246	},
247	)
248
249	if not obj:	×
250	return [status, import_params]	×
251
252	if obj.import_status == models.ImportJob.ImportStatus.CREATED:	×
253	return [status, import_params]	×
254
255	if obj.import_status in models.ImportJob.results_statuses:	×
256	return [status, result, data, import_params]	×
257
258	if obj.import_status in models.ImportJob.progress_statuses:	×
259	return [status, progress, import_params]	×
260
261	return [status, traceback_, import_params]	×
262
263	@admin.action(description="Cancel selected jobs")	12✔
264	def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):	12✔
265	"""Admin action for cancelling data import."""
266	for job in queryset:	×
267	try:	×
268	job.cancel_import()	×
269	self.message_user(	×
270	request,
271	_(f"Import of {job} canceled"),
272	messages.SUCCESS,
273	)
274	except ValueError as error:	×
275	self.message_user(request, str(error), messages.ERROR)	×
276
277	@admin.action(description="Confirm selected jobs")	12✔
278	def confirm_jobs(self, request: WSGIRequest, queryset: QuerySet):	12✔
279	"""Admin action for confirming data import."""
280	for job in queryset:	×
281	try:	×
282	job.confirm_import()	×
283	self.message_user(	×
284	request,
285	_(f"Import of {job} confirmed"),
286	messages.SUCCESS,
287	)
288	except ValueError as error:	×
289	self.message_user(request, str(error), messages.ERROR)	×
290
291
292	admin.site.register(models.ImportJob, ImportJobAdmin)	12✔

saritasa-nest / django-import-export-extensions / 6144128305

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous