6143269168

Committed 11 Sep 2023 07:26AM UTC coverage: 78.27% (+0.4%) from 77.914%

Build # 6143269168

Build Type

Pull #15

github

Committed by

web-flow

Commit Message

Merge e508d664d into ca9e3dc6a

Pull Request Pull Request #15: Make possible to pass resource kwargs in view sets

Run Details

22 of 22 new or added lines in 4 files covered. (100.0%)

35 existing lines in 2 files now uncovered.

1095 of 1399 relevant lines covered (78.27%)

9.38 hits per line

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

80.77

/import_export_extensions/admin/mixins/export_mixin.py

import typing

from django.core.exceptions import PermissionDenied
from django.core.handlers.wsgi import WSGIRequest
from django.http import (
    HttpResponse,
    HttpResponseForbidden,
    HttpResponseRedirect,
)
from django.shortcuts import get_object_or_404
from django.template.response import TemplateResponse
from django.urls import re_path, reverse
from django.utils.translation import gettext_lazy as _

from import_export import admin as base_admin
from import_export import forms as base_forms
from import_export import mixins as base_mixins

from ... import models
from . import types


class CeleryExportAdminMixin(
    base_mixins.BaseExportMixin,
    base_admin.ImportExportMixinBase,
):
    """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"

    # 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"

    import_export_change_list_template = "admin/import_export/change_list_export.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 = base_admin.ExportMixin.has_export_permission

    @property
    def model_info(self) -> types.ModelInfo:
        """Get info of exported model."""
        return types.ModelInfo(
            meta=self.model._meta,
        )

    def get_export_resource_kwargs(self, request, *args, **kwargs):
        """Provide escape settings to resource kwargs."""
        kwargs = super().get_export_resource_kwargs(request, *args, **kwargs)
        kwargs.update({
            "escape_output": self.should_escape_output,
            "escape_html": self.should_escape_html,
            "escape_formulae": self.should_escape_formulae,
        })
        return kwargs

    def get_context_data(
        self,
        request: WSGIRequest,
        **kwargs,
    ) -> dict[str, typing.Any]:
        """Get context data."""
        return {}

    def get_urls(self):
        """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, *args, **kwargs):
        """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()
        form = base_forms.ExportForm(
            formats,
            request.POST or None,
            resources=self.get_export_resource_classes(),
        )
        resource_kwargs = self.get_export_resource_kwargs(
            request=request,
            *args,
            **kwargs,
        )
        if request.method == "POST" and form.is_valid():
            file_format = formats[int(form.cleaned_data["file_format"])]
            # 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),
                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_context_data(request)
        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_context_data(request)
        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_context_data(request)

        if request.method != "GET":
            return HttpResponseForbidden()

        # 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,
            file_format_path=".".join(
                [
                    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 _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=dict(job_id=job.id))
        query = request.GET.urlencode()
        if query:
            url = f"{url}?{query}"
        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=dict(job_id=job.id))
        query = request.GET.urlencode()
        if query:
            url = f"{url}?{query}"
        return HttpResponseRedirect(redirect_to=url)

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

1	import typing	12✔
2
3	from django.core.exceptions import PermissionDenied	12✔
4	from django.core.handlers.wsgi import WSGIRequest	12✔
5	from django.http import (	12✔
6	HttpResponse,
7	HttpResponseForbidden,
8	HttpResponseRedirect,
9	)
10	from django.shortcuts import get_object_or_404	12✔
11	from django.template.response import TemplateResponse	12✔
12	from django.urls import re_path, reverse	12✔
13	from django.utils.translation import gettext_lazy as _	12✔
14
15	from import_export import admin as base_admin	12✔
16	from import_export import forms as base_forms	12✔
17	from import_export import mixins as base_mixins	12✔
18
19	from ... import models	12✔
20	from . import types	12✔
21
22
23	class CeleryExportAdminMixin(	12✔
24	base_mixins.BaseExportMixin,
25	base_admin.ImportExportMixinBase,
26	):
27	"""Admin mixin for celery export.
28
29	Admin export work-flow is:
30	GET `celery_export_action()` - display form with format type input
31
32	POST `celery_export_action()` - create ExportJob and starts data export
33	This view redirects to next view:
34
35	GET `celery_export_job_status_view()` - display ExportJob status (with
36	progress bar). When data exporting is done, redirect to next view:
37
38	GET `celery_export_job_results_view()` - display export results. If no
39	errors - success message and link to the file with exported data.
40	If errors - traceback and error message.
41
42	"""
43	# export data encoding
44	to_encoding = "utf-8"	12✔
45
46	# template used to display ExportForm
47	celery_export_template_name = "admin/import_export/export.html"	12✔
48
49	export_status_template_name = "admin/import_export_extensions/celery_export_status.html"	12✔
50
51	export_results_template_name = "admin/import_export_extensions/celery_export_results.html"	12✔
52
53	import_export_change_list_template = "admin/import_export/change_list_export.html"	12✔
54
55	import_export_change_list_template = "admin/import_export/change_list_export.html"	12✔
56
57	import_export_change_list_template = "admin/import_export/change_list_export.html"	12✔
58
59	# Statuses that should be displayed on 'results' page
60	export_results_statuses = models.ExportJob.export_finished_statuses	12✔
61
62	# Copy methods of mixin from original package to reuse it here
63	has_export_permission = base_admin.ExportMixin.has_export_permission	12✔
64
65	@property	12✔
66	def model_info(self) -> types.ModelInfo:	12✔
67	"""Get info of exported model."""
UNCOV 68	return types.ModelInfo(	×
69	meta=self.model._meta,
70	)
71
72	def get_export_resource_kwargs(self, request, args, *kwargs):	12✔
73	"""Provide escape settings to resource kwargs."""
74	kwargs = super().get_export_resource_kwargs(request, args, *kwargs)	12✔
75	kwargs.update({	12✔
76	"escape_output": self.should_escape_output,
77	"escape_html": self.should_escape_html,
78	"escape_formulae": self.should_escape_formulae,
79	})
80	return kwargs	12✔
81
82	def get_context_data(	12✔
83	self,
84	request: WSGIRequest,
85	**kwargs,
86	) -> dict[str, typing.Any]:
87	"""Get context data."""
UNCOV 88	return {}	×
89
90	def get_urls(self):	12✔
91	"""Return list of urls.
92
93	/<model/celery-export/:
94	ExportForm ('export_action' method)
95	/<model>/celery-export/<ID>/:
96	status of ExportJob and progress bar ('export_job_status_view')
97	/<model>/celery-export/<ID>/results/:
98	table with export results (errors)
99
100	"""
101	urls = super().get_urls()	12✔
102	export_urls = [	12✔
103	re_path(
104	r"^celery-export/$",
105	self.admin_site.admin_view(self.celery_export_action),
106	name=f"{self.model_info.app_model_name}_export",
107	),
108	re_path(
109	r"^celery-export/(?P<job_id>\d+)/$",
110	self.admin_site.admin_view(self.export_job_status_view),
111	name=(
112	f"{self.model_info.app_model_name}"
113	f"_export_job_status"
114	),
115	),
116	re_path(
117	r"^celery-export/(?P<job_id>\d+)/results/$",
118	self.admin_site.admin_view(
119	self.export_job_results_view,
120	),
121	name=(
122	f"{self.model_info.app_model_name}"
123	f"_export_job_results"
124	),
125	),
126	]
127	return export_urls + urls	12✔
128
129	def celery_export_action(self, request, args, *kwargs):	12✔
130	"""Show and handle export.
131
132	GET: show export form with format_type input
133	POST: create ExportJob instance and redirect to it's status
134
135	"""
136	if not self.has_export_permission(request):	12✔
UNCOV 137	raise PermissionDenied	×
138
139	formats = self.get_export_formats()	12✔
140	form = base_forms.ExportForm(	12✔
141	formats,
142	request.POST or None,
143	resources=self.get_export_resource_classes(),
144	)
145	resource_kwargs = self.get_export_resource_kwargs(	12✔
146	request=request,
147	*args,
148	**kwargs,
149	)
150	if request.method == "POST" and form.is_valid():	12✔
151	file_format = formats[int(form.cleaned_data["file_format"])]	12✔
152	# create ExportJob and redirect to page with it's status
153	job = self.create_export_job(	12✔
154	request=request,
155	resource_class=self.choose_export_resource_class(form),
156	resource_kwargs=resource_kwargs,
157	file_format=file_format,
158	)
159	return self._redirect_to_export_status_page(	12✔
160	request=request,
161	job=job,
162	)
163
164	# GET: display Export Form
165	context = self.get_context_data(request)	12✔
166	context.update(self.admin_site.each_context(request))	12✔
167
168	context["title"] = _("Export")	12✔
169	context["form"] = form	12✔
170	context["opts"] = self.model_info.meta	12✔
171	request.current_app = self.admin_site.name	12✔
172
173	return TemplateResponse(	12✔
174	request=request,
175	template=[self.celery_export_template_name],
176	context=context,
177	)
178
179	def export_job_status_view(	12✔
180	self,
181	request: WSGIRequest,
182	job_id: int,
183	**kwargs,
184	) -> HttpResponse:
185	"""View to track export job status.
186
187	Displays current export job status and progress (using JS + another
188	view).
189
190	If job result is ready - redirects to another page to see results.
191
192	"""
193	if not self.has_export_permission(request):	12✔
UNCOV 194	raise PermissionDenied	×
195
196	job = self.get_export_job(request=request, job_id=job_id)	12✔
197	if job.export_status in self.export_results_statuses:	12✔
198	return self._redirect_to_export_results_page(	12✔
199	request=request,
200	job=job,
201	)
202
203	context = self.get_context_data(request)	×
204	job_url = reverse("admin:export_job_progress", args=(job.id,))	×
205
206	context["title"] = _("Export status")	×
207	context["opts"] = self.model_info.meta	×
UNCOV 208	context["export_job"] = job	×
UNCOV 209	context["export_job_url"] = job_url	×
UNCOV 210	request.current_app = self.admin_site.name	×
UNCOV 211	return TemplateResponse(	×
212	request=request,
213	template=[self.export_status_template_name],
214	context=context,
215	)
216
217	def export_job_results_view(	12✔
218	self,
219	request: WSGIRequest,
220	job_id: int,
221	*args,
222	**kwargs,
223	) -> HttpResponse:
224	"""Display export results.
225
226	GET-request:
227	* show message
228	* if no errors - show file link
229	* if errors - show traceback and error
230
231	"""
232	if not self.has_export_permission(request):	12✔
233	raise PermissionDenied	×
234
235	job = self.get_export_job(request=request, job_id=job_id)	12✔
236	if job.export_status not in self.export_results_statuses:	12✔
UNCOV 237	return self._redirect_to_export_status_page(	×
238	request=request,
239	job=job,
240	)
241
242	context = self.get_context_data(request)	12✔
243
244	if request.method != "GET":	12✔
UNCOV 245	return HttpResponseForbidden()	×
246
247	# GET request, show export results
248	context["title"] = _("Export results")	12✔
249	context["opts"] = self.model._meta	12✔
250	context["export_job"] = job	12✔
251	context["result"] = job.export_status	12✔
252
253	return TemplateResponse(	12✔
254	request=request,
255	template=[self.export_results_template_name],
256	context=context,
257	)
258
259	def create_export_job(	12✔
260	self,
261	request: WSGIRequest,
262	resource_class: types.ResourceType,
263	resource_kwargs: dict[str, typing.Any],
264	file_format: types.FormatType,
265	) -> models.ExportJob:
266	"""Create and return instance of export job with chosen format."""
267	job = models.ExportJob.objects.create(	12✔
268	resource_path=resource_class.class_path,
269	resource_kwargs=resource_kwargs,
270	file_format_path=".".join(
271	[
272	file_format.__module__,
273	file_format.__name__,
274	],
275	),
276	)
277	return job	12✔
278
279	def get_export_job(	12✔
280	self,
281	request: WSGIRequest,
282	job_id: int,
283	) -> models.ExportJob:
284	"""Get ExportJob instance.
285
286	Raises:
287	Http404
288
289	"""
290	return get_object_or_404(models.ExportJob, id=job_id)	12✔
291
292	def _redirect_to_export_status_page(	12✔
293	self,
294	request: WSGIRequest,
295	job: models.ExportJob,
296	) -> HttpResponse:
297	"""Shortcut for redirecting to job's status page."""
298	url_name = (	12✔
299	f"admin:{self.model_info.app_model_name}_export_job_status"
300	)
301	url = reverse(url_name, kwargs=dict(job_id=job.id))	12✔
302	query = request.GET.urlencode()	12✔
303	if query:	12✔
UNCOV 304	url = f"{url}?{query}"	×
305	return HttpResponseRedirect(redirect_to=url)	12✔
306
307	def _redirect_to_export_results_page(	12✔
308	self,
309	request: WSGIRequest,
310	job: models.ExportJob,
311	) -> HttpResponse:
312	"""Shortcut for redirecting to job's results page."""
313	url_name = (	12✔
314	f"admin:{self.model_info.app_model_name}_export_job_results"
315	)
316	url = reverse(url_name, kwargs=dict(job_id=job.id))	12✔
317	query = request.GET.urlencode()	12✔
318	if query:	12✔
UNCOV 319	url = f"{url}?{query}"	×
320	return HttpResponseRedirect(redirect_to=url)	12✔
321
322	def changelist_view(	12✔
323	self,
324	request: WSGIRequest,
325	context: typing.Optional[dict[str, typing.Any]] = None,
326	):
327	"""Add the check for permission to changelist template context."""
UNCOV 328	context = context or {}	×
UNCOV 329	context["has_export_permission"] = True	×
UNCOV 330	return super().changelist_view(request, context)	×

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

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