6943835407

Committed 21 Nov 2023 12:32PM UTC coverage: 78.492% (+0.4%) from 78.121%

Build # 6943835407

Build Type

Pull #25

github

Committed by

web-flow

Commit Message

Merge 2c3246b3f into a3aa8ce29

Pull Request Pull Request #25: Feature/force import

Run Details

46 of 55 new or added lines in 7 files covered. (83.64%)

26 existing lines in 3 files now uncovered.

1135 of 1446 relevant lines covered (78.49%)

9.41 hits per line

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

82.17

/import_export_extensions/admin/mixins/import_mixin.py

import typing

from django.conf import settings
from django.core.cache import cache
from django.core.exceptions import PermissionDenied
from django.core.handlers.wsgi import WSGIRequest
from django.forms.forms import Form
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 ..forms import ExtendedImportForm
from . import types


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

    # 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 = base_admin.ImportMixin.generate_log_entries
    get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log
    has_import_permission = base_admin.ImportMixin.has_import_permission

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

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

    def get_import_context_data(self, **kwargs):
        """Get context for import data."""
        return self.get_context_data(**kwargs)

    def get_urls(self):
        """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,
    ):
        """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_context_data(request)
        resource_classes = self.get_import_resource_classes()

        form = ExtendedImportForm(
            self.get_import_formats(),
            request.POST or None,
            request.FILES or None,
            resources=resource_classes,
        )
        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)
            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
        resources = [
            resource_class(**resource_kwargs)
            for resource_class in resource_classes
        ]

        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"] = [
            (
                resource.get_display_name(),
                [f.column_name for f in resource.get_user_visible_fields()],
            )
            for resource in resources
        ]

        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_context_data(request)
        job_url = reverse("admin:import_job_progress", args=(job.id,))
        context.update(
            dict(
                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_context_data(request=request)

        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:
                # display import form
                context["import_form"] = base_forms.ImportForm(
                    import_formats=self.get_import_formats(),
                )
            else:
                context["confirm_form"] = Form()

            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 data is invalid - error
        if job.import_status != models.ImportJob.ImportStatus.PARSED:
            return HttpResponseForbidden(
                "Data invalid, before importing data "
                "needs to be successfully parsed."
                f"Current status: {job.import_status}",
            )

        # start celery task for data importing
        job.confirm_import()
        return self._redirect_to_import_status_page(request=request, job=job)

    def create_import_job(
        self,
        request: WSGIRequest,
        form: Form,
        resource: types.ResourceObj,
    ):
        """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["force_import"],
        )

    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=dict(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=dict(job_id=job.id))
        query = request.GET.urlencode()
        url = f"{url}?{query}" if query else url
        if not job.import_status == models.ImportJob.ImportStatus.PARSED:
            return HttpResponseRedirect(redirect_to=url)

        # Redirections add one by one links to `redirect_to`
        key = request.session.get("redirect_key", None)
        session = request.session
        if key:
            links = cache.get(key)
            try:
                session["redirect_to"] = links[0]
                del links[0]
                cache.set(key, links)
            except (TypeError, IndexError):
                session.pop("redirect_to", None)
                session.pop("redirect_key", None)
                cache.delete(key)

        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_import_permission"] = self.has_import_permission(request)
        return super().changelist_view(request, context)

1	import typing	12✔
2
3	from django.conf import settings	12✔
4	from django.core.cache import cache	12✔
5	from django.core.exceptions import PermissionDenied	12✔
6	from django.core.handlers.wsgi import WSGIRequest	12✔
7	from django.forms.forms import Form	12✔
8	from django.http import (	12✔
9	HttpResponse,
10	HttpResponseForbidden,
11	HttpResponseRedirect,
12	)
13	from django.shortcuts import get_object_or_404	12✔
14	from django.template.response import TemplateResponse	12✔
15	from django.urls import re_path, reverse	12✔
16	from django.utils.translation import gettext_lazy as _	12✔
17
18	from import_export import admin as base_admin	12✔
19	from import_export import forms as base_forms	12✔
20	from import_export import mixins as base_mixins	12✔
21
22	from ... import models	12✔
23	from ..forms import ExtendedImportForm	12✔
24	from . import types	12✔
25
26
27	class CeleryImportAdminMixin(	12✔
28	base_mixins.BaseImportMixin,
29	base_admin.ImportExportMixinBase,
30	):
31	"""Admin mixin for celery import.
32
33	Admin import work-flow is:
34
35	GET ``celery_import_action()`` - display form with import file input
36
37	POST ``celery_import_action()`` - save file and create ImportJob.
38	This view redirects to next view:
39
40	GET ``celery_import_job_status_view()`` - display ImportJob status (with
41	progress bar and critical errors occurred). When data parsing is
42	done, redirect to next view:
43
44	GET ``celery_import_job_results_view()`` - display rows that will be
45	imported and data parse errors. If no errors - next step.
46	If errors - display same form as in ``import_action()``
47
48	POST ``celery_import_job_results_view()`` - start data importing and
49	redirect back to GET ``celery_import_job_status_view()``
50	with progress bar and import totals.
51
52	"""
53	# Import data encoding
54	from_encoding = "utf-8"	12✔
55
56	# Statuses that should be displayed on 'results' page
57	results_statuses = models.ImportJob.results_statuses	12✔
58
59	# Template used to display ImportForm
60	celery_import_template = "admin/import_export/import.html"	12✔
61
62	# Template used to display status of import jobs
63	import_status_template = "admin/import_export_extensions/celery_import_status.html"	12✔
64
65	# template used to display results of import jobs
66	import_result_template_name = "admin/import_export_extensions/celery_import_results.html"	12✔
67
68	import_export_change_list_template = "admin/import_export/change_list_import.html"	12✔
69
70	skip_admin_log = None	12✔
71	# Copy methods of mixin from original package to reuse it here
72	generate_log_entries = base_admin.ImportMixin.generate_log_entries	12✔
73	get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log	12✔
74	has_import_permission = base_admin.ImportMixin.has_import_permission	12✔
75
76	@property	12✔
77	def model_info(self) -> types.ModelInfo:	12✔
78	"""Get info of imported model."""
79	return types.ModelInfo(	12✔
80	meta=self.model._meta,
81	)
82
83	def get_context_data(	12✔
84	self,
85	request: WSGIRequest,
86	**kwargs,
87	) -> dict[str, typing.Any]:
88	"""Get context data."""
89	return {}	12✔
90
91	def get_import_context_data(self, **kwargs):	12✔
92	"""Get context for import data."""
UNCOV 93	return self.get_context_data(**kwargs)	×
94
95	def get_urls(self):	12✔
96	"""Return list of urls.
97
98	* /<model>/<celery-import>/:
99	ImportForm ('celery_import_action' method)
100	* /<model>/<celery-import>/<ID>/:
101	status of ImportJob and progress bar
102	('celery_import_job_status_view')
103	* /<model>/<celery-import>/<ID>/results/:
104	table with import results (errors) and import confirmation
105	('celery_import_job_results_view')
106
107	"""
108	urls = super().get_urls()	12✔
109	import_urls = [	12✔
110	re_path(
111	r"^celery-import/$",
112	self.admin_site.admin_view(self.celery_import_action),
113	name=f"{self.model_info.app_model_name}_import",
114	),
115	re_path(
116	r"^celery-import/(?P<job_id>\d+)/$",
117	self.admin_site.admin_view(self.celery_import_job_status_view),
118	name=(
119	f"{self.model_info.app_model_name}"
120	f"_import_job_status"
121	),
122	),
123	re_path(
124	r"^celery-import/(?P<job_id>\d+)/results/$",
125	self.admin_site.admin_view(
126	self.celery_import_job_results_view,
127	),
128	name=(
129	f"{self.model_info.app_model_name}"
130	f"_import_job_results"
131	),
132	),
133	]
134	return import_urls + urls	12✔
135
136	def celery_import_action(	12✔
137	self,
138	request: WSGIRequest,
139	*args,
140	**kwargs,
141	):
142	"""Show and handle ImportForm.
143
144	GET:
145	show import form with data_file input form
146	POST:
147	create ImportJob instance and redirect to it's status
148
149	"""
150	if not self.has_import_permission(request):	12✔
UNCOV 151	raise PermissionDenied	×
152
153	context = self.get_context_data(request)	12✔
154	resource_classes = self.get_import_resource_classes()	12✔
155
156	form = ExtendedImportForm(	12✔
157	self.get_import_formats(),
158	request.POST or None,
159	request.FILES or None,
160	resources=resource_classes,
161	)
162	resource_kwargs = self.get_import_resource_kwargs(request)	12✔
163
164	if request.method == "POST" and form.is_valid():	12✔
165	# create ImportJob and redirect to page with it's status
166	resource_class = self.choose_import_resource_class(form)	12✔
167	job = self.create_import_job(	12✔
168	request=request,
169	resource=resource_class(**resource_kwargs),
170	form=form,
171	)
172	return self._redirect_to_import_status_page(	12✔
173	request=request,
174	job=job,
175	)
176
177	# GET: display Import Form
178	resources = [	12✔
179	resource_class(**resource_kwargs)
180	for resource_class in resource_classes
181	]
182
183	context.update(self.admin_site.each_context(request))	12✔
184
185	context["title"] = _("Import")	12✔
186	context["form"] = form	12✔
187	context["opts"] = self.model_info.meta	12✔
188	context["media"] = self.media + form.media	12✔
189	context["fields_list"] = [	12✔
190	(
191	resource.get_display_name(),
192	[f.column_name for f in resource.get_user_visible_fields()],
193	)
194	for resource in resources
195	]
196
197	request.current_app = self.admin_site.name	12✔
198	return TemplateResponse(	12✔
199	request,
200	[self.celery_import_template],
201	context,
202	)
203
204	def celery_import_job_status_view(	12✔
205	self,
206	request: WSGIRequest,
207	job_id: int,
208	**kwargs,
209	) -> HttpResponse:
210	"""View to track import job status.
211
212	Displays current import job status and progress (using JS + another
213	view).
214
215	If job result is ready - redirects to another page to see results.
216
217	Also generates admin log entries if the job has `IMPORTED` status.
218
219	"""
220	if not self.has_import_permission(request):	12✔
UNCOV 221	raise PermissionDenied	×
222
223	job = self.get_import_job(request, job_id)	12✔
224	if job.import_status in self.results_statuses:	12✔
225	if job.import_status == models.ImportJob.ImportStatus.IMPORTED:	12✔
226	self.generate_log_entries(job.result, request)	12✔
227	return self._redirect_to_results_page(	12✔
228	request=request,
229	job=job,
230	)
231
232	context = self.get_context_data(request)	×
233	job_url = reverse("admin:import_job_progress", args=(job.id,))	×
UNCOV 234	context.update(	×
235	dict(
236	title=_("Import status"),
237	opts=self.model_info.meta,
238	import_job=job,
239	import_job_url=job_url,
240	),
241	)
242	request.current_app = self.admin_site.name	×
UNCOV 243	return TemplateResponse(	×
244	request=request,
245	template=[self.import_status_template],
246	context=context,
247	)
248
249	def celery_import_job_results_view(	12✔
250	self,
251	request: WSGIRequest,
252	job_id: int,
253	*args,
254	**kwargs,
255	) -> HttpResponse:
256	"""Display table with import results and import confirm form.
257
258	GET-request:
259	* show row results
260	* if data valid - show import confirmation form
261	* if data invalid - show ImportForm for uploading other file
262
263	POST-request:
264	* start data importing if data is correct
265
266	"""
267	if not self.has_import_permission(request):	12✔
UNCOV 268	raise PermissionDenied	×
269
270	job = self.get_import_job(request=request, job_id=job_id)	12✔
271	if job.import_status not in self.results_statuses:	12✔
UNCOV 272	return self._redirect_to_import_status_page(	×
273	request=request,
274	job=job,
275	)
276
277	context = self.get_context_data(request=request)	12✔
278
279	if request.method == "GET":	12✔
280	# GET request, show parse results
281	result = job.result	12✔
282	context["import_job"] = job	12✔
283	context["result"] = result	12✔
284	context["title"] = _("Import results")	12✔
285
286	if job.import_status != models.ImportJob.ImportStatus.PARSED:	12✔
287	# display import form
288	context["import_form"] = base_forms.ImportForm(	12✔
289	import_formats=self.get_import_formats(),
290	)
291	else:
292	context["confirm_form"] = Form()	12✔
293
294	context.update(self.admin_site.each_context(request))	12✔
295	context["opts"] = self.model_info.meta	12✔
296	request.current_app = self.admin_site.name	12✔
297	return TemplateResponse(	12✔
298	request,
299	[self.import_result_template_name],
300	context,
301	)
302
303	# POST request. If data is invalid - error
304	if job.import_status != models.ImportJob.ImportStatus.PARSED:	12✔
UNCOV 305	return HttpResponseForbidden(	×
306	"Data invalid, before importing data "
307	"needs to be successfully parsed."
308	f"Current status: {job.import_status}",
309	)
310
311	# start celery task for data importing
312	job.confirm_import()	12✔
313	return self._redirect_to_import_status_page(request=request, job=job)	12✔
314
315	def create_import_job(	12✔
316	self,
317	request: WSGIRequest,
318	form: Form,
319	resource: types.ResourceObj,
320	):
321	"""Create and return instance of import job."""
322	return models.ImportJob.objects.create(	12✔
323	resource_path=resource.class_path,
324	data_file=form.cleaned_data["import_file"],
325	resource_kwargs=resource.resource_init_kwargs,
326	created_by=request.user,
327	skip_parse_step=getattr(settings, "IMPORT_EXPORT_SKIP_ADMIN_CONFIRM", False),
328	force_import=form.cleaned_data["force_import"],
329	)
330
331	def get_import_job(	12✔
332	self,
333	request: WSGIRequest,
334	job_id: int,
335	) -> models.ImportJob:
336	"""Get ImportJob instance.
337
338	Raises:
339	Http404
340
341	"""
342	return get_object_or_404(klass=models.ImportJob, id=job_id)	12✔
343
344	def _redirect_to_import_status_page(	12✔
345	self,
346	request: WSGIRequest,
347	job: models.ImportJob,
348	) -> HttpResponseRedirect:
349	"""Shortcut for redirecting to job's status page."""
350	url_name = (	12✔
351	f"admin:{self.model_info.app_model_name}_import_job_status"
352	)
353	url = reverse(url_name, kwargs=dict(job_id=job.id))	12✔
354	query = request.GET.urlencode()	12✔
355	url = f"{url}?{query}" if query else url	12✔
356	return HttpResponseRedirect(redirect_to=url)	12✔
357
358	def _redirect_to_results_page(	12✔
359	self,
360	request: WSGIRequest,
361	job: models.ImportJob,
362	) -> HttpResponseRedirect:
363	"""Shortcut for redirecting to job's results page."""
364	url_name = (	12✔
365	f"admin:{self.model_info.app_model_name}_import_job_results"
366	)
367	url = reverse(url_name, kwargs=dict(job_id=job.id))	12✔
368	query = request.GET.urlencode()	12✔
369	url = f"{url}?{query}" if query else url	12✔
370	if not job.import_status == models.ImportJob.ImportStatus.PARSED:	12✔
371	return HttpResponseRedirect(redirect_to=url)	12✔
372
373	# Redirections add one by one links to `redirect_to`
374	key = request.session.get("redirect_key", None)	12✔
375	session = request.session	12✔
376	if key:	12✔
377	links = cache.get(key)	×
378	try:	×
379	session["redirect_to"] = links[0]	×
380	del links[0]	×
381	cache.set(key, links)	×
382	except (TypeError, IndexError):	×
383	session.pop("redirect_to", None)	×
384	session.pop("redirect_key", None)	×
UNCOV 385	cache.delete(key)	×
386
387	return HttpResponseRedirect(redirect_to=url)	12✔
388
389	def changelist_view(	12✔
390	self,
391	request: WSGIRequest,
392	context: typing.Optional[dict[str, typing.Any]] = None,
393	):
394	"""Add the check for permission to changelist template context."""
395	context = context or {}	×
396	context["has_import_permission"] = self.has_import_permission(request)	×
UNCOV 397	return super().changelist_view(request, context)	×

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

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