• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

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

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

Pull #15

github

web-flow
Merge e508d664d into ca9e3dc6a
Pull Request #15: Make possible to pass resource kwargs in view sets

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

83.33
/import_export_extensions/admin/mixins/import_mixin.py
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 . import types
12✔
24

25

26
class CeleryImportAdminMixin(
12✔
27
    base_mixins.BaseImportMixin,
28
    base_admin.ImportExportMixinBase,
29
):
30
    """Admin mixin for celery import.
31

32
    Admin import work-flow is:
33

34
    GET ``celery_import_action()`` - display form with import file input
35

36
    POST ``celery_import_action()`` - save file and create ImportJob.
37
        This view redirects to next view:
38

39
    GET ``celery_import_job_status_view()`` - display ImportJob status (with
40
        progress bar and critical errors occurred). When data parsing is
41
        done, redirect to next view:
42

43
    GET ``celery_import_job_results_view()`` - display rows that will be
44
        imported and data parse errors. If no errors - next step.
45
        If errors - display same form as in ``import_action()``
46

47
    POST ``celery_import_job_results_view()`` - start data importing and
48
        redirect back to GET ``celery_import_job_status_view()``
49
        with progress bar and import totals.
50

51
    """
52
    # Import data encoding
53
    from_encoding = "utf-8"
12✔
54

55
    # Statuses that should be displayed on 'results' page
56
    results_statuses = models.ImportJob.results_statuses
12✔
57

58
    # Template used to display ImportForm
59
    celery_import_template = "admin/import_export/import.html"
12✔
60

61
    # Template used to display status of import jobs
62
    import_status_template = "admin/import_export_extensions/celery_import_status.html"
12✔
63

64
    # template used to display results of import jobs
65
    import_result_template_name = "admin/import_export_extensions/celery_import_results.html"
12✔
66

67
    import_export_change_list_template = "admin/import_export/change_list_import.html"
12✔
68

69
    skip_admin_log = None
12✔
70
    # Copy methods of mixin from original package to reuse it here
71
    generate_log_entries = base_admin.ImportMixin.generate_log_entries
12✔
72
    get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log
12✔
73
    has_import_permission = base_admin.ImportMixin.has_import_permission
12✔
74

75
    import_export_change_list_template = "admin/import_export/change_list_import.html"
12✔
76

77
    skip_admin_log = None
12✔
78
    # Copy methods of mixin from original package to reuse it here
79
    generate_log_entries = base_admin.ImportMixin.generate_log_entries
12✔
80
    get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log
12✔
81
    has_import_permission = base_admin.ImportMixin.has_import_permission
12✔
82

83
    import_export_change_list_template = "admin/import_export/change_list_import.html"
12✔
84

85
    skip_admin_log = None
12✔
86
    # Copy methods of mixin from original package to reuse it here
87
    generate_log_entries = base_admin.ImportMixin.generate_log_entries
12✔
88
    get_skip_admin_log = base_admin.ImportMixin.get_skip_admin_log
12✔
89
    has_import_permission = base_admin.ImportMixin.has_import_permission
12✔
90

91
    @property
12✔
92
    def model_info(self) -> types.ModelInfo:
12✔
93
        """Get info of imported model."""
94
        return types.ModelInfo(
12✔
95
            meta=self.model._meta,
96
        )
97

98
    def get_context_data(
12✔
99
        self,
100
        request: WSGIRequest,
101
        **kwargs,
102
    ) -> dict[str, typing.Any]:
103
        """Get context data."""
104
        return {}
12✔
105

106
    def get_import_context_data(self, **kwargs):
12✔
107
        """Get context for import data."""
UNCOV
108
        return self.get_context_data(**kwargs)
×
109

110
    def get_urls(self):
12✔
111
        """Return list of urls.
112

113
        * /<model>/<celery-import>/:
114
            ImportForm ('celery_import_action' method)
115
        * /<model>/<celery-import>/<ID>/:
116
            status of ImportJob and progress bar
117
            ('celery_import_job_status_view')
118
        * /<model>/<celery-import>/<ID>/results/:
119
            table with import results (errors) and import confirmation
120
            ('celery_import_job_results_view')
121

122
        """
123
        urls = super().get_urls()
12✔
124
        import_urls = [
12✔
125
            re_path(
126
                r"^celery-import/$",
127
                self.admin_site.admin_view(self.celery_import_action),
128
                name=f"{self.model_info.app_model_name}_import",
129
            ),
130
            re_path(
131
                r"^celery-import/(?P<job_id>\d+)/$",
132
                self.admin_site.admin_view(self.celery_import_job_status_view),
133
                name=(
134
                    f"{self.model_info.app_model_name}"
135
                    f"_import_job_status"
136
                ),
137
            ),
138
            re_path(
139
                r"^celery-import/(?P<job_id>\d+)/results/$",
140
                self.admin_site.admin_view(
141
                    self.celery_import_job_results_view,
142
                ),
143
                name=(
144
                    f"{self.model_info.app_model_name}"
145
                    f"_import_job_results"
146
                ),
147
            ),
148
        ]
149
        return import_urls + urls
12✔
150

151
    def celery_import_action(
12✔
152
        self,
153
        request: WSGIRequest,
154
        *args,
155
        **kwargs,
156
    ):
157
        """Show and handle ImportForm.
158

159
        GET:
160
            show import form with data_file input form
161
        POST:
162
            create ImportJob instance and redirect to it's status
163

164
        """
165
        if not self.has_import_permission(request):
12✔
UNCOV
166
            raise PermissionDenied
×
167

168
        context = self.get_context_data(request)
12✔
169
        resource_classes = self.get_import_resource_classes()
12✔
170

171
        form = base_forms.ImportForm(
12✔
172
            self.get_import_formats(),
173
            request.POST or None,
174
            request.FILES or None,
175
            resources=resource_classes,
176
        )
177
        resource_kwargs = self.get_import_resource_kwargs(request)
12✔
178

179
        if request.method == "POST" and form.is_valid():
12✔
180
            # create ImportJob and redirect to page with it's status
181
            resource_class = self.choose_import_resource_class(form)
12✔
182
            job = self.create_import_job(
12✔
183
                request=request,
184
                resource=resource_class(**resource_kwargs),
185
                form=form,
186
            )
187
            return self._redirect_to_import_status_page(
12✔
188
                request=request,
189
                job=job,
190
            )
191

192
        # GET: display Import Form
193
        resources = [
12✔
194
            resource_class(**resource_kwargs)
195
            for resource_class in resource_classes
196
        ]
197

198
        context.update(self.admin_site.each_context(request))
12✔
199

200
        context["title"] = _("Import")
12✔
201
        context["form"] = form
12✔
202
        context["opts"] = self.model_info.meta
12✔
203
        context["media"] = self.media + form.media
12✔
204
        context["fields_list"] = [
12✔
205
            (
206
                resource.get_display_name(),
207
                [f.column_name for f in resource.get_user_visible_fields()],
208
            )
209
            for resource in resources
210
        ]
211

212
        request.current_app = self.admin_site.name
12✔
213
        return TemplateResponse(
12✔
214
            request,
215
            [self.celery_import_template],
216
            context,
217
        )
218

219
    def celery_import_job_status_view(
12✔
220
        self,
221
        request: WSGIRequest,
222
        job_id: int,
223
        **kwargs,
224
    ) -> HttpResponse:
225
        """View to track import job status.
226

227
        Displays current import job status and progress (using JS + another
228
        view).
229

230
        If job result is ready - redirects to another page to see results.
231

232
        Also generates admin log entries if the job has `IMPORTED` status.
233

234
        """
235
        if not self.has_import_permission(request):
12✔
UNCOV
236
            raise PermissionDenied
×
237

238
        job = self.get_import_job(request, job_id)
12✔
239
        if job.import_status in self.results_statuses:
12✔
240
            if job.import_status == models.ImportJob.ImportStatus.IMPORTED:
12✔
241
                self.generate_log_entries(job.result, request)
12✔
242
            return self._redirect_to_results_page(
12✔
243
                request=request,
244
                job=job,
245
            )
246

UNCOV
247
        context = self.get_context_data(request)
×
UNCOV
248
        job_url = reverse("admin:import_job_progress", args=(job.id,))
×
UNCOV
249
        context.update(
×
250
            dict(
251
                title=_("Import status"),
252
                opts=self.model_info.meta,
253
                import_job=job,
254
                import_job_url=job_url,
255
            ),
256
        )
UNCOV
257
        request.current_app = self.admin_site.name
×
UNCOV
258
        return TemplateResponse(
×
259
            request=request,
260
            template=[self.import_status_template],
261
            context=context,
262
        )
263

264
    def celery_import_job_results_view(
12✔
265
        self,
266
        request: WSGIRequest,
267
        job_id: int,
268
        *args,
269
        **kwargs,
270
    ) -> HttpResponse:
271
        """Display table with import results and import confirm form.
272

273
        GET-request:
274
            * show row results
275
            * if data valid - show import confirmation form
276
            * if data invalid - show ImportForm for uploading other file
277

278
        POST-request:
279
            * start data importing if data is correct
280

281
        """
282
        if not self.has_import_permission(request):
12✔
UNCOV
283
            raise PermissionDenied
×
284

285
        job = self.get_import_job(request=request, job_id=job_id)
12✔
286
        if job.import_status not in self.results_statuses:
12✔
UNCOV
287
            return self._redirect_to_import_status_page(
×
288
                request=request,
289
                job=job,
290
            )
291

292
        context = self.get_context_data(request=request)
12✔
293

294
        if request.method == "GET":
12✔
295
            # GET request, show parse results
296
            result = job.result
12✔
297
            context["import_job"] = job
12✔
298
            context["result"] = result
12✔
299
            context["title"] = _("Import results")
12✔
300

301
            if job.import_status != models.ImportJob.ImportStatus.PARSED:
12✔
302
                # display import form
303
                context["import_form"] = base_forms.ImportForm(
12✔
304
                    import_formats=job.resource.SUPPORTED_FORMATS,
305
                )
306
            else:
307
                context["confirm_form"] = Form()
12✔
308

309
            context.update(self.admin_site.each_context(request))
12✔
310
            context["opts"] = self.model_info.meta
12✔
311
            request.current_app = self.admin_site.name
12✔
312
            return TemplateResponse(
12✔
313
                request,
314
                [self.import_result_template_name],
315
                context,
316
            )
317

318
        # POST request. If data is invalid - error
319
        if job.import_status != models.ImportJob.ImportStatus.PARSED:
12✔
UNCOV
320
            return HttpResponseForbidden(
×
321
                "Data invalid, before importing data "
322
                "needs to be successfully parsed."
323
                f"Current status: {job.import_status}",
324
            )
325

326
        # start celery task for data importing
327
        job.confirm_import()
12✔
328
        return self._redirect_to_import_status_page(request=request, job=job)
12✔
329

330
    def create_import_job(
12✔
331
        self,
332
        request: WSGIRequest,
333
        form: Form,
334
        resource: types.ResourceObj,
335
    ):
336
        """Create and return instance of import job."""
337
        return models.ImportJob.objects.create(
12✔
338
            resource_path=resource.class_path,
339
            data_file=form.cleaned_data["import_file"],
340
            resource_kwargs=resource.resource_init_kwargs,
341
            created_by=request.user,
342
            skip_parse_step=getattr(settings, "IMPORT_EXPORT_SKIP_ADMIN_CONFIRM", False),
343
        )
344

345
    def get_import_job(
12✔
346
        self,
347
        request: WSGIRequest,
348
        job_id: int,
349
    ) -> models.ImportJob:
350
        """Get ImportJob instance.
351

352
        Raises:
353
            Http404
354

355
        """
356
        return get_object_or_404(klass=models.ImportJob, id=job_id)
12✔
357

358
    def _redirect_to_import_status_page(
12✔
359
        self,
360
        request: WSGIRequest,
361
        job: models.ImportJob,
362
    ) -> HttpResponseRedirect:
363
        """Shortcut for redirecting to job's status page."""
364
        url_name = (
12✔
365
            f"admin:{self.model_info.app_model_name}_import_job_status"
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
        return HttpResponseRedirect(redirect_to=url)
12✔
371

372
    def _redirect_to_results_page(
12✔
373
        self,
374
        request: WSGIRequest,
375
        job: models.ImportJob,
376
    ) -> HttpResponseRedirect:
377
        """Shortcut for redirecting to job's results page."""
378
        url_name = (
12✔
379
            f"admin:{self.model_info.app_model_name}_import_job_results"
380
        )
381
        url = reverse(url_name, kwargs=dict(job_id=job.id))
12✔
382
        query = request.GET.urlencode()
12✔
383
        url = f"{url}?{query}" if query else url
12✔
384
        if not job.import_status == models.ImportJob.ImportStatus.PARSED:
12✔
385
            return HttpResponseRedirect(redirect_to=url)
12✔
386

387
        # Redirections add one by one links to `redirect_to`
388
        key = request.session.get("redirect_key", None)
12✔
389
        session = request.session
12✔
390
        if key:
12✔
UNCOV
391
            links = cache.get(key)
×
UNCOV
392
            try:
×
393
                session["redirect_to"] = links[0]
×
394
                del links[0]
×
395
                cache.set(key, links)
×
UNCOV
396
            except (TypeError, IndexError):
×
UNCOV
397
                session.pop("redirect_to", None)
×
UNCOV
398
                session.pop("redirect_key", None)
×
UNCOV
399
                cache.delete(key)
×
400

401
        return HttpResponseRedirect(redirect_to=url)
12✔
402

403
    def changelist_view(
12✔
404
        self,
405
        request: WSGIRequest,
406
        context: typing.Optional[dict[str, typing.Any]] = None,
407
    ):
408
        """Add the check for permission to changelist template context."""
UNCOV
409
        context = context or {}
×
UNCOV
410
        context["has_import_permission"] = self.has_import_permission(request)
×
UNCOV
411
        return super().changelist_view(request, context)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc