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

43.66

/import_export_extensions/admin/model_admins/export_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.urls import re_path
from django.utils.translation import gettext_lazy as _

from ... import models, tasks
from .. import forms
from . import mixins


class ExportJobAdmin(
    mixins.BaseImportExportJobAdminMixin,
    admin.ModelAdmin,
):
    """Admin class for debugging ExportJob."""

    form = forms.ExportJobAdminForm
    exclude = ("result",)
    list_display = (
        "id",
        "export_status",
        "_model",
        "export_started",
        "export_finished",
        "created",
        "created_by",
    )
    list_display_links = (
        "id",
        "export_status",
        "_model",
    )
    export_job_model = models.ExportJob
    list_filter = ("export_status",)
    list_select_related = ("created_by",)
    actions = (
        "cancel_jobs",
    )

    def export_data_action(
        self,
        request: WSGIRequest,
        obj: models.ExportJob,
    ):
        """Admin action for starting data export.

        Data export should auto start after export confirmation. But there
        may be issues with celery and task did not start. So this action
        for such cases.

        """
        tasks.export_data_task.delay(obj.id)

    export_data_action.label = _("Start Export")

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

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

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

        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, obj=None):
        """Return readonly fields.

        Some fields are editable for new ExportJob.

        """
        readonly_fields = [
            "export_status",
            "traceback",
            "file_format_path",
            "created",
            "export_started",
            "export_finished",
            "error_message",
            "_model",
        ]
        if obj:
            readonly_fields.extend(
                [
                    "resource_path",
                    "data_file",
                    "resource_kwargs",
                ],
            )

        return readonly_fields

    def get_fieldsets(
        self,
        request: WSGIRequest,
        obj: typing.Optional[models.ExportJob] = None,
    ):
        """Get fieldsets depending on object status."""
        status = (
            _("Status"),
            {
                "fields": (
                    "export_status",
                    "_model",
                    "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 (
            not obj
            or 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):
        """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)


admin.site.register(models.ExportJob, ExportJobAdmin)

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.urls import re_path	12✔
8	from django.utils.translation import gettext_lazy as _	12✔
9
10	from ... import models, tasks	12✔
11	from .. import forms	12✔
12	from . import mixins	12✔
13
14
15	class ExportJobAdmin(	12✔
16	mixins.BaseImportExportJobAdminMixin,
17	admin.ModelAdmin,
18	):
19	"""Admin class for debugging ExportJob."""
20
21	form = forms.ExportJobAdminForm	12✔
22	exclude = ("result",)	12✔
23	list_display = (	12✔
24	"id",
25	"export_status",
26	"_model",
27	"export_started",
28	"export_finished",
29	"created",
30	"created_by",
31	)
32	list_display_links = (	12✔
33	"id",
34	"export_status",
35	"_model",
36	)
37	export_job_model = models.ExportJob	12✔
38	list_filter = ("export_status",)	12✔
39	list_select_related = ("created_by",)	12✔
40	actions = (	12✔
41	"cancel_jobs",
42	)
43
44	def export_data_action(	12✔
45	self,
46	request: WSGIRequest,
47	obj: models.ExportJob,
48	):
49	"""Admin action for starting data export.
50
51	Data export should auto start after export confirmation. But there
52	may be issues with celery and task did not start. So this action
53	for such cases.
54
55	"""
56	tasks.export_data_task.delay(obj.id)	×
57
58	export_data_action.label = _("Start Export")	12✔
59
60	def get_urls(self):	12✔
61	"""Add url to get current export job progress in JSON representation.
62
63	/admin/import_export_extensions/exportjob/<job_id>/progress/
64
65	"""
66	urls = super().get_urls()	12✔
67	export_urls = [	12✔
68	re_path(
69	route=r"^(?P<job_id>\d+)/progress/$",
70	view=self.admin_site.admin_view(self.export_job_progress_view),
71	name="export_job_progress",
72	),
73	]
74	return export_urls + urls	12✔
75
76	def export_job_progress_view(	12✔
77	self,
78	request: WSGIRequest,
79	job_id: int,
80	**kwargs,
81	):
82	"""View to return `ExportJob` status as JSON.
83
84	If current status is exporting, view also returns job state
85	and percent of completed work.
86
87	Return:
88	Response: dictionary with status (optionally, state and percent).
89
90	"""
91	try:	×
92	job: models.ExportJob = self.export_job_model.objects.get(	×
93	id=job_id,
94	)
95	except self.export_job_model.DoesNotExist as error:	×
96	return JsonResponse(dict(validation_error=error.args[0]))	×
97
98	response_data = dict(status=job.export_status.title())	×
99
100	if job.export_status != models.ExportJob.ExportStatus.EXPORTING:	×
101	return JsonResponse(response_data)	×
102
103	percent = 0	×
104	total = 0	×
105	current = 0	×
106	info = job.progress["info"]	×
107
108	if info and info["total"]:	×
109	percent = int(100 / info["total"] * info["current"])	×
110	total = info["total"]	×
111	current = info["current"]	×
112
113	response_data.update(	×
114	dict(
115	state=job.progress["state"],
116	percent=percent,
117	total=total,
118	current=current,
119	),
120	)
121	return JsonResponse(response_data)	×
122
123	def get_readonly_fields(self, request, obj=None):	12✔
124	"""Return readonly fields.
125
126	Some fields are editable for new ExportJob.
127
128	"""
129	readonly_fields = [	×
130	"export_status",
131	"traceback",
132	"file_format_path",
133	"created",
134	"export_started",
135	"export_finished",
136	"error_message",
137	"_model",
138	]
139	if obj:	×
140	readonly_fields.extend(	×
141	[
142	"resource_path",
143	"data_file",
144	"resource_kwargs",
145	],
146	)
147
148	return readonly_fields	×
149
150	def get_fieldsets(	12✔
151	self,
152	request: WSGIRequest,
153	obj: typing.Optional[models.ExportJob] = None,
154	):
155	"""Get fieldsets depending on object status."""
156	status = (	×
157	_("Status"),
158	{
159	"fields": (
160	"export_status",
161	"_model",
162	"created",
163	"export_started",
164	"export_finished",
165	),
166	},
167	)
168	progress = (	×
169	_("Status"),
170	{
171	"fields": (
172	"export_status",
173	"export_progressbar",
174	),
175	},
176	)
177	export_params = (	×
178	_("Export params"),
179	{
180	"fields": (
181	"resource_path",
182	"resource_kwargs",
183	"file_format_path",
184	),
185	"classes": ("collapse",),
186	},
187	)
188	traceback_fields = (	×
189	_("Traceback"),
190	{
191	"fields": (
192	"error_message",
193	"traceback",
194	),
195	},
196	)
197	result = (	×
198	_("Export results"),
199	{
200	"fields": ("data_file",),
201	},
202	)
203
204	if (	×
205	not obj
206	or obj.export_status == models.ExportJob.ExportStatus.CREATED
207	):
208	return [status, export_params]	×
209
210	if obj.export_status == models.ExportJob.ExportStatus.EXPORTED:	×
211	return [status, result, export_params]	×
212
213	if obj.export_status == models.ExportJob.ExportStatus.EXPORTING:	×
214	return [progress, export_params]	×
215
216	return [status, traceback_fields, export_params]	×
217
218	@admin.action(description="Cancel selected jobs")	12✔
219	def cancel_jobs(self, request: WSGIRequest, queryset: QuerySet):	12✔
220	"""Admin action for cancelling data export."""
221	for job in queryset:	×
222	try:	×
223	job.cancel_export()	×
224	self.message_user(	×
225	request,
226	_(f"Export of {job} canceled"),
227	messages.SUCCESS,
228	)
229	except ValueError as error:	×
230	self.message_user(request, str(error), messages.ERROR)	×
231
232
233	admin.site.register(models.ExportJob, ExportJobAdmin)	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