8700791003

Committed 16 Apr 2024 06:10AM UTC coverage: 80.809%. Remained the same

Build # 8700791003

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #32 from saritasa-nest/dependabot/pip/sqlparse-0.5.0

Run Details

1238 of 1532 relevant lines covered (80.81%)

8.08 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",
                    "input_errors_file",
                    "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_file",
                    "_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	10✔
2
3	from django.contrib import admin, messages	10✔
4	from django.core.handlers.wsgi import WSGIRequest	10✔
5	from django.db.models import QuerySet	10✔
6	from django.http import JsonResponse	10✔
7	from django.template.loader import render_to_string	10✔
8	from django.urls import re_path	10✔
9	from django.utils.translation import gettext_lazy as _	10✔
10
11	from ... import models	10✔
12	from .. import forms	10✔
13	from . import mixins	10✔
14
15
16	class ImportJobAdmin(	10✔
17	mixins.BaseImportExportJobAdminMixin,
18	admin.ModelAdmin,
19	):
20	"""Admin class for debugging ImportJob."""
21
22	form = forms.ImportJobAdminForm	10✔
23	exclude = ("result",)	10✔
24	list_display = (	10✔
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 = (	10✔
35	"id",
36	"import_status",
37	"_model",
38	)
39	import_job_model = models.ImportJob	10✔
40	list_filter = ("import_status",)	10✔
41	list_select_related = ("created_by",)	10✔
42	actions = (	10✔
43	"cancel_jobs",
44	"confirm_jobs",
45	)
46
47	def get_queryset(self, request: WSGIRequest):	10✔
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):	10✔
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()	10✔
63	import_urls = [	10✔
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	10✔
71
72	def import_job_progress_view(	10✔
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(	10✔
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	"input_errors_file",
145	"data_file",
146	"resource_kwargs",
147	],
148	)
149
150	return readonly_fields	×
151
152	def _show_results(	10✔
153	self,
154	obj: typing.Optional[models.ImportJob] = None,
155	) -> str:
156	"""Show results totals.
157
158	Example return value:
159	New: 99
160	Update: 1
161	Delete: 0
162	Skip: 0
163	Error: 0
164
165	"""
166	if not obj:	×
167	return ""	×
168
169	result_sections = []	×
170	for key, value in obj.result.totals.items():	×
171	status_template = f"{key.title()}: {value}"	×
172	result_sections.append(status_template)	×
173
174	return "\n".join(result_sections)	×
175
176	_show_results.short_description = _("Parse/Import results")	10✔
177
178	def _input_errors(self, job: models.ImportJob):	10✔
179	"""Render html with input errors."""
180	template = "admin/import_export_extensions/import_job_results.html"	×
181	return render_to_string(	×
182	template,
183	dict(result=job.result),
184	)
185
186	_input_errors.short_description = "Import data"	10✔
187	_input_errors.allow_tags = True	10✔
188
189	def get_fieldsets(	10✔
190	self,
191	request: WSGIRequest,
192	obj: typing.Optional[models.ImportJob] = None,
193	):
194	"""Get fieldsets depending on object status."""
195	status = (	×
196	_("Status"),
197	{
198	"fields": (
199	"import_status",
200	"_model",
201	"created_by",
202	"created",
203	"parse_finished",
204	"import_started",
205	"import_finished",
206	),
207	},
208	)
209	progress = (	×
210	_("Status"),
211	{
212	"fields": (
213	"import_status",
214	"import_progressbar",
215	),
216	},
217	)
218	import_params = (	×
219	_("Import params"),
220	{
221	"fields": (
222	"data_file",
223	"resource_path",
224	"resource_kwargs",
225	),
226	"classes": ("collapse",),
227	},
228	)
229	traceback_ = (	×
230	_("Traceback"),
231	{
232	"fields": ("traceback",),
233	},
234	)
235	result = (	×
236	_("Result totals"),
237	{
238	"fields": ("_show_results",),
239	"classes": ("collapse",),
240	},
241	)
242	data = (	×
243	_("Importing data"),
244	{
245	"fields": (
246	"input_errors_file",
247	"_input_errors",
248	),
249	"classes": ("collapse",),
250	},
251	)
252
253	if not obj:	×
254	return [status, import_params]	×
255
256	if obj.import_status == models.ImportJob.ImportStatus.CREATED:	×
257	return [status, import_params]	×
258
259	if obj.import_status in models.ImportJob.results_statuses:	×
260	return [status, result, data, import_params]	×
261
262	if obj.import_status in models.ImportJob.progress_statuses:	×
263	return [status, progress, import_params]	×
264
265	return [status, traceback_, import_params]	×
266
267	@admin.action(description="Cancel selected jobs")	10✔
268	def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):	10✔
269	"""Admin action for cancelling data import."""
270	for job in queryset:	×
271	try:	×
272	job.cancel_import()	×
273	self.message_user(	×
274	request,
275	_(f"Import of {job} canceled"),
276	messages.SUCCESS,
277	)
278	except ValueError as error:	×
279	self.message_user(request, str(error), messages.ERROR)	×
280
281	@admin.action(description="Confirm selected jobs")	10✔
282	def confirm_jobs(self, request: WSGIRequest, queryset: QuerySet):	10✔
283	"""Admin action for confirming data import."""
284	for job in queryset:	×
285	try:	×
286	job.confirm_import()	×
287	self.message_user(	×
288	request,
289	_(f"Import of {job} confirmed"),
290	messages.SUCCESS,
291	)
292	except ValueError as error:	×
293	self.message_user(request, str(error), messages.ERROR)	×
294
295
296	admin.site.register(models.ImportJob, ImportJobAdmin)	10✔

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

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