diff --git a/docs/installation.rst b/docs/installation.rst index 0d13d60..073c103 100644 --- a/docs/installation.rst +++ b/docs/installation.rst @@ -103,3 +103,6 @@ Settings from django-import-export There are also available `settings from original django-import-export `_ package. + +Only `IMPORT_EXPORT_TMP_STORAGE_CLASS` setting does not affect anything, because the storage +is not used in `CeleryImportAdminMixin` implementation. diff --git a/import_export_extensions/admin/mixins/export_mixin.py b/import_export_extensions/admin/mixins/export_mixin.py index 8d6f630..04d7ea9 100644 --- a/import_export_extensions/admin/mixins/export_mixin.py +++ b/import_export_extensions/admin/mixins/export_mixin.py @@ -1,7 +1,5 @@ import typing -from django.conf import settings -from django.contrib.auth import get_permission_codename from django.core.exceptions import PermissionDenied from django.core.handlers.wsgi import WSGIRequest from django.http import ( @@ -52,9 +50,14 @@ class CeleryExportAdminMixin( export_results_template_name = "admin/import_export_ext/celery_export_results.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.""" @@ -62,18 +65,15 @@ def model_info(self) -> types.ModelInfo: meta=self.model._meta, ) - def get_export_data( - self, - resource: types.ResourceObj, - file_format: types.FormatType, - queryset, - *args, - **kwargs, - ): - """Return file_format representation for given queryset.""" - data = resource.export(queryset, *args, **kwargs) - export_data = file_format().export_data(data) - return export_data + 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, @@ -315,20 +315,6 @@ def _redirect_to_export_results_page( url = f"{url}?{query}" return HttpResponseRedirect(redirect_to=url) - def has_export_permission(self, request): - """Return whether a request has export permission.""" - EXPORT_PERMISSION_CODE = getattr( - settings, - "IMPORT_EXPORT_EXPORT_PERMISSION_CODE", - None, - ) - if EXPORT_PERMISSION_CODE is None: - return True - - opts = self.opts - codename = get_permission_codename(EXPORT_PERMISSION_CODE, opts) - return request.user.has_perm("%s.%s" % (opts.app_label, codename)) - def changelist_view(self, request, context=None): """Add the check for permission to changelist template context.""" context = context or {} diff --git a/import_export_extensions/admin/mixins/import_mixin.py b/import_export_extensions/admin/mixins/import_mixin.py index e2b1950..e307862 100644 --- a/import_export_extensions/admin/mixins/import_mixin.py +++ b/import_export_extensions/admin/mixins/import_mixin.py @@ -1,7 +1,6 @@ import typing from django.conf import settings -from django.contrib.auth import get_permission_codename from django.core.cache import cache from django.core.exceptions import PermissionDenied from django.core.handlers.wsgi import WSGIRequest @@ -65,6 +64,14 @@ class CeleryImportAdminMixin( # template used to display results of import jobs import_result_template_name = "admin/import_export_ext/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.""" @@ -206,12 +213,16 @@ def celery_import_job_status_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, @@ -312,6 +323,7 @@ def create_import_job( 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), ) def get_import_job( @@ -372,20 +384,6 @@ def _redirect_to_results_page( return HttpResponseRedirect(redirect_to=url) - def has_import_permission(self, request): - """Return whether a request has import permission.""" - IMPORT_PERMISSION_CODE = getattr( - settings, - "IMPORT_EXPORT_IMPORT_PERMISSION_CODE", - None, - ) - if IMPORT_PERMISSION_CODE is None: - return True - - opts = self.opts - codename = get_permission_codename(IMPORT_PERMISSION_CODE, opts) - return request.user.has_perm("%s.%s" % (opts.app_label, codename)) - def changelist_view(self, request, context=None): """Add the check for permission to changelist template context.""" context = context or {} diff --git a/import_export_extensions/migrations/0003_importjob_skip_parse_step.py b/import_export_extensions/migrations/0003_importjob_skip_parse_step.py new file mode 100644 index 0000000..f93c3b3 --- /dev/null +++ b/import_export_extensions/migrations/0003_importjob_skip_parse_step.py @@ -0,0 +1,18 @@ +# Generated by Django 4.2.4 on 2023-08-26 04:28 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('import_export_extensions', '0002_alter_exportjob_export_status'), + ] + + operations = [ + migrations.AddField( + model_name='importjob', + name='skip_parse_step', + field=models.BooleanField(default=False, help_text='Start importing without confirmation', verbose_name='Skip parse step'), + ), + ] diff --git a/import_export_extensions/models/export_job.py b/import_export_extensions/models/export_job.py index 8fa2722..4802ca0 100644 --- a/import_export_extensions/models/export_job.py +++ b/import_export_extensions/models/export_job.py @@ -336,6 +336,9 @@ def _export_data_inner(self): # file object (formats inherited from `BaseZipExport`) export_data = self.file_format.export_data( dataset=self.result, + escape_output=self.resource_kwargs.get("escape_output", False), + escape_html=self.resource_kwargs.get("escape_html", False), + escape_formulae=self.resource_kwargs.get("escape_formulae", False), ) # create file if `export_data` is not file if not hasattr(export_data, "read"): diff --git a/import_export_extensions/models/import_job.py b/import_export_extensions/models/import_job.py index d2aa550..9c56039 100644 --- a/import_export_extensions/models/import_job.py +++ b/import_export_extensions/models/import_job.py @@ -216,6 +216,12 @@ class ImportStatus(models.TextChoices): help_text=_("User which started import"), ) + skip_parse_step = models.BooleanField( + default=False, + help_text=_("Start importing without confirmation"), + verbose_name=_("Skip parse step"), + ) + class Meta: verbose_name = _("Import job") verbose_name_plural = _("Import jobs") @@ -247,7 +253,20 @@ def save( using=using, update_fields=update_fields, ) - if is_created: + if not is_created: + return + + if self.skip_parse_step: + self.import_task_id = str(uuid.uuid4()) + self.import_started = timezone.now() + self.save( + update_fields=[ + "import_task_id", + "import_started", + ], + ) + transaction.on_commit(self._start_import_data_task) + else: self.parse_task_id = str(uuid.uuid4()) self.save(update_fields=["parse_task_id"]) transaction.on_commit(self.start_parse_data_task) @@ -423,8 +442,13 @@ def _start_import_data_task(self): def import_data(self): """Import data from `data_file` to DB.""" + expected_status = ( + self.ImportStatus.CREATED + if self.skip_parse_step + else self.ImportStatus.CONFIRMED + ) self._check_import_status_correctness( - expected_statuses=(self.ImportStatus.CONFIRMED,), + expected_statuses=(expected_status,), ) self.import_status = self.ImportStatus.IMPORTING