Models

Bases: BaseJob

Abstract model for managing celery export jobs.

Encapsulate all logic related to celery export.

Export steps:

1. Create ExportJob with resource initialization parameters.
2. Try to export all data to file.
3. If everything is correct - export data to file from database.

Export file saves in media files.

Source code in import_export_extensions/models/export_job.py

class ExportJob(BaseJob):
    """Abstract model for managing celery export jobs.

    Encapsulate all logic related to celery export.

    Export steps:

    1. Create ``ExportJob`` with resource initialization parameters.
    2. Try to export all data to file.
    3. If everything is correct - export data to file from database.

    Export file saves in media files.

    """

    class ExportStatus(models.TextChoices):
        """ExportJob possible statuses.

        * CREATED:
            export job is just created, no exporting done
        * EXPORTING:
            export job started
        * EXPORT_ERROR:
            DB queryset not exported, errors
        * EXPORTED:
            DB queryset exported, no errors occurred

        State diagrams:

            CREATED
               |
            EXPORTING  - (EXPORT_ERROR)
               |
            EXPORTED

        """

        CREATED = "CREATED", _("Created")
        EXPORTING = "EXPORTING", _("Exporting")
        EXPORT_ERROR = "EXPORT_ERROR", _("Export Error")
        EXPORTED = "EXPORTED", _("Exported")
        CANCELLED = "CANCELLED", _("Cancelled")

    export_finished_statuses = (
        ExportStatus.EXPORTED,
        ExportStatus.EXPORT_ERROR,
    )

    export_status = models.CharField(
        max_length=20,
        choices=ExportStatus.choices,
        default=ExportStatus.CREATED,
        verbose_name=_("Job status"),
    )

    file_format_path = models.CharField(
        max_length=128,
        verbose_name=_("Export path to class"),
        help_text=_(
            "Export file format as path to base_formats class",
        ),
    )

    data_file = models.FileField(
        max_length=512,
        storage=tools.select_storage,
        verbose_name=_("Data file"),
        upload_to=tools.upload_export_file_to,
        help_text=_("File that contain exported data"),
    )

    export_task_id = models.CharField(  # noqa: DJ001
        verbose_name=_("Export task ID"),
        max_length=36,
        null=True,
        blank=True,
        help_text=_("Celery task ID that start `export_data`"),
    )

    export_started = models.DateTimeField(
        verbose_name=_("Export started"),
        editable=False,
        blank=True,
        null=True,
    )

    export_finished = models.DateTimeField(
        verbose_name=_("Export finished"),
        editable=False,
        blank=True,
        null=True,
    )

    class Meta:
        verbose_name = _("Export job")
        verbose_name_plural = _("Export jobs")

    def __str__(self) -> str:
        """Return string representation."""
        resource_name = pathlib.Path(self.resource_path).suffix.lstrip(".")
        file_format = pathlib.Path(self.file_format_path).suffix.lstrip(".")

        return f"ExportJob(resource={resource_name}, {file_format=})"

    def save(
        self,
        force_insert: bool = False,
        force_update: bool = False,
        using: str | None = None,
        update_fields: collections.abc.Iterable[str] | None = None,
    ) -> None:
        """Start task for data exporting when ExportJob is created.

        Celery task is manually called with `apply_async`, to provide
        possibility of custom `task_id` with which task will be run.

        """
        is_created = self._state.adding
        super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )
        if is_created:
            self.export_task_id = str(uuid.uuid4())
            self.save(update_fields=["export_task_id"])
            transaction.on_commit(self._start_export_data_task)

    @property
    def file_format(self) -> base_formats.Format:
        """Get initialized format instance."""
        return module_loading.import_string(self.file_format_path)()

    @property
    def export_filename(self) -> str:
        """Get filename for export result file."""
        return self.resource.generate_export_filename(
            file_format=self.file_format,
        ).replace("/", "-")

    @property
    def progress(self) -> TaskStateInfo | None:
        """Return dict with export state."""
        if (
            self.export_task_id
            and self.export_status == self.ExportStatus.EXPORTING
        ):
            return self._get_task_state(self.export_task_id)

        return None

    def _check_export_status_correctness(
        self,
        expected_statuses: collections.abc.Sequence[ExportStatus],
    ) -> None:
        """Raise `ValueError` if `ExportJob` is in incorrect state."""
        if self.export_status not in expected_statuses:
            raise ValueError(
                f"ExportJob with id {self.id} has incorrect status: "
                f"`{self.export_status}`. Expected statuses:"
                f" {[status.value for status in expected_statuses]}",
            )

    def _start_export_data_task(self) -> None:
        """Start export data task."""
        from .. import tasks

        tasks.export_data_task.apply_async(
            kwargs={"job_id": self.pk},
            task_id=self.export_task_id,
        )

    def export_data(self) -> None:
        """Export data to `data_file` from DB."""
        self.export_status = self.ExportStatus.EXPORTING
        self.export_started = timezone.now()
        self.save(
            update_fields=[
                "export_status",
                "export_started",
            ],
        )
        try:
            self._export_data_inner()
            self.export_status = self.ExportStatus.EXPORTED
            self.export_finished = timezone.now()
            self.save(
                update_fields=[
                    "export_status",
                    "export_finished",
                ],
            )
        except Exception as error:
            self._handle_error(
                error_message=str(error),
                traceback=traceback.format_exc(),
                exception=error,
            )

    def cancel_export(self) -> None:
        """Cancel current data export.

        ExportJob can be CANCELLED only from following states:
            - CREATED
            - EXPORTING

        """
        self._check_export_status_correctness(
            expected_statuses=(  # type: ignore
                self.ExportStatus.CREATED,
                self.ExportStatus.EXPORTING,
            ),
        )

        # send signal to celery to revoke task
        current_app.control.revoke(self.export_task_id, terminate=True)

        self.export_status = self.ExportStatus.CANCELLED
        self.save(update_fields=["export_status"])

    def _export_data_inner(self) -> None:
        """Run export process with saving to file."""
        self.result = self.resource.export(**self.resource_kwargs)
        self.save(update_fields=["result"])

        # `export_data` may be bytes (base formats such as xlsx, csv, etc.) or
        # file object (formats inherited from `BaseZipExport`)
        export_data = self.file_format.export_data(
            dataset=self.result,
            **self.resource.get_export_data_format_kwargs(
                file_format=self.file_format,
            ),
        )
        # create file if `export_data` is not file
        if not hasattr(export_data, "read"):
            export_data = django_files.base.ContentFile(
                content=encoding.force_bytes(export_data),
            )
        self.data_file.save(
            name=self.export_filename,
            content=export_data,
            save=True,
        )

    def _get_task_state(self, task_id: str) -> TaskStateInfo:
        """Get state info for passed task_id.

        This method may change job status if task failed, but we did not
        save info about this to DB.

        This may happen if task start exporting, but processes was killed.
        In that case, job have status `importing`, but it can't be finished.

        """
        async_result = result.AsyncResult(task_id)
        if async_result.state not in states.EXCEPTION_STATES:
            return {
                "state": async_result.state,
                "info": async_result.info,
            }

        self._handle_error(
            error_message=str(async_result.info),
            traceback=str(async_result.traceback),
        )
        return {
            "state": async_result.state,
            "info": {},
        }

    def _handle_error(
        self,
        error_message: str,
        traceback: str,
        exception: Exception | None = None,
    ) -> None:
        """Update job's status in case of error."""
        self.export_status = self.ExportStatus.EXPORT_ERROR
        error_message_limit = self._meta.get_field("error_message").max_length
        self.error_message = error_message[:error_message_limit]
        self.traceback = traceback
        self.save(
            update_fields=[
                "error_message",
                "traceback",
                "export_status",
            ],
        )
        signals.export_job_failed.send(
            sender=self.__class__,
            instance=self,
            error_message=self.error_message,
            traceback=self.traceback,
            exception=exception,
        )

export_filename property

Get filename for export result file.

file_format property

Get initialized format instance.

progress property

Return dict with export state.

ExportStatus

Bases: TextChoices

ExportJob possible statuses.

• CREATED: export job is just created, no exporting done
• EXPORTING: export job started
• EXPORT_ERROR: DB queryset not exported, errors
• EXPORTED: DB queryset exported, no errors occurred

State diagrams:

CREATED
   |
EXPORTING  - (EXPORT_ERROR)
   |
EXPORTED

Source code in import_export_extensions/models/export_job.py

class ExportStatus(models.TextChoices):
    """ExportJob possible statuses.

    * CREATED:
        export job is just created, no exporting done
    * EXPORTING:
        export job started
    * EXPORT_ERROR:
        DB queryset not exported, errors
    * EXPORTED:
        DB queryset exported, no errors occurred

    State diagrams:

        CREATED
           |
        EXPORTING  - (EXPORT_ERROR)
           |
        EXPORTED

    """

    CREATED = "CREATED", _("Created")
    EXPORTING = "EXPORTING", _("Exporting")
    EXPORT_ERROR = "EXPORT_ERROR", _("Export Error")
    EXPORTED = "EXPORTED", _("Exported")
    CANCELLED = "CANCELLED", _("Cancelled")

__str__()

Return string representation.

Source code in import_export_extensions/models/export_job.py

def __str__(self) -> str:
    """Return string representation."""
    resource_name = pathlib.Path(self.resource_path).suffix.lstrip(".")
    file_format = pathlib.Path(self.file_format_path).suffix.lstrip(".")

    return f"ExportJob(resource={resource_name}, {file_format=})"

cancel_export()

Cancel current data export.

ExportJob can be CANCELLED only from following states

• CREATED
• EXPORTING

Source code in import_export_extensions/models/export_job.py

def cancel_export(self) -> None:
    """Cancel current data export.

    ExportJob can be CANCELLED only from following states:
        - CREATED
        - EXPORTING

    """
    self._check_export_status_correctness(
        expected_statuses=(  # type: ignore
            self.ExportStatus.CREATED,
            self.ExportStatus.EXPORTING,
        ),
    )

    # send signal to celery to revoke task
    current_app.control.revoke(self.export_task_id, terminate=True)

    self.export_status = self.ExportStatus.CANCELLED
    self.save(update_fields=["export_status"])

export_data()

Export data to data_file from DB.

Source code in import_export_extensions/models/export_job.py

def export_data(self) -> None:
    """Export data to `data_file` from DB."""
    self.export_status = self.ExportStatus.EXPORTING
    self.export_started = timezone.now()
    self.save(
        update_fields=[
            "export_status",
            "export_started",
        ],
    )
    try:
        self._export_data_inner()
        self.export_status = self.ExportStatus.EXPORTED
        self.export_finished = timezone.now()
        self.save(
            update_fields=[
                "export_status",
                "export_finished",
            ],
        )
    except Exception as error:
        self._handle_error(
            error_message=str(error),
            traceback=traceback.format_exc(),
            exception=error,
        )

save(force_insert=False, force_update=False, using=None, update_fields=None)

Start task for data exporting when ExportJob is created.

Celery task is manually called with apply_async, to provide possibility of custom task_id with which task will be run.

Source code in import_export_extensions/models/export_job.py

def save(
    self,
    force_insert: bool = False,
    force_update: bool = False,
    using: str | None = None,
    update_fields: collections.abc.Iterable[str] | None = None,
) -> None:
    """Start task for data exporting when ExportJob is created.

    Celery task is manually called with `apply_async`, to provide
    possibility of custom `task_id` with which task will be run.

    """
    is_created = self._state.adding
    super().save(
        force_insert=force_insert,
        force_update=force_update,
        using=using,
        update_fields=update_fields,
    )
    if is_created:
        self.export_task_id = str(uuid.uuid4())
        self.save(update_fields=["export_task_id"])
        transaction.on_commit(self._start_export_data_task)

Bases: BaseJob

Abstract model for managing celery import jobs.

Encapsulate all logic related to celery import.

Import steps:

1. Create ImportJob with resource initialization parameters and file with data to be imported from.
2. Dry run. Try to import all data from file and collect statistics ( errors, new rows, updated rows).
3. If data for import is correct - import data from file to database.

Source code in import_export_extensions/models/import_job.py

class ImportJob(BaseJob):
    """Abstract model for managing celery import jobs.

    Encapsulate all logic related to celery import.

    Import steps:

    1. Create ``ImportJob`` with resource initialization parameters and file
        with data to be imported from.
    2. Dry run. Try to import all data from file and collect statistics (
        errors, new rows, updated rows).
    3. If data for import is correct - import data from file to database.

    """

    class ImportStatus(models.TextChoices):
        """ImportJob possible statuses.

        * CREATED:
            import job is just created, no parsing done
        * PARSING:
            parse job started
        * PARSED:
            data_file parsed, no errors in data occurred
        * INPUT_ERROR:
            data_file parsed, data contains errors
        * PARSE_ERROR:
            data_file can't be parsed (invalid format, etc.)
        * IMPORT_CONFIRMED
            import confirmed but not started yet
        * IMPORTING:
            importing job started
        * IMPORTED:
            data from data_file imported to DB w/o errors
        * IMPORT_ERROR:
            unknown error during import
        * CANCELLED:
            import job has been cancelled (revoked)

        State diagrams:

            CREATED
               |
            .parse_data()
               |
            PARSING  - (INPUT_ERROR, PARSE_ERROR)
               |
            PARSED
               |
            .confirm_import()
               |
            IMPORT_CONFIRMED
               |
            .import_data()
               |
            IMPORTING - IMPORT_ERROR
               |
            IMPORTED

        """

        CREATED = "CREATED", _("Created")
        PARSING = "PARSING", _("Parsing")
        PARSED = "PARSED", _("Parsed")
        INPUT_ERROR = "INPUT_ERROR", _("Input data error")
        PARSE_ERROR = "PARSE_ERROR", _("Parse error")
        CONFIRMED = "CONFIRMED", _("Import confirmed")
        IMPORTING = "IMPORTING", _("Importing")
        IMPORTED = "IMPORTED", _("Imported")
        IMPORT_ERROR = "IMPORT_ERROR", _("Import error")
        CANCELLED = "CANCELLED", _("Cancelled")

    results_statuses = (
        ImportStatus.PARSED,
        ImportStatus.INPUT_ERROR,
        ImportStatus.IMPORTED,
        ImportStatus.IMPORT_ERROR,
    )

    progress_statuses = (
        ImportStatus.PARSING,
        ImportStatus.IMPORTING,
    )

    parse_finished_statuses = (
        ImportStatus.INPUT_ERROR,
        ImportStatus.PARSE_ERROR,
        ImportStatus.PARSED,
    )

    import_finished_statuses = (
        ImportStatus.IMPORTED,
        ImportStatus.IMPORT_ERROR,
    )

    success_statuses = (
        ImportStatus.IMPORTED,
        ImportStatus.PARSED,
    )

    failure_statuses = (
        ImportStatus.INPUT_ERROR,
        ImportStatus.PARSE_ERROR,
    )

    import_status = models.CharField(
        max_length=20,
        choices=ImportStatus.choices,
        default=ImportStatus.CREATED,
        verbose_name=_("Job status"),
    )

    data_file = models.FileField(
        max_length=512,
        storage=tools.select_storage,
        verbose_name=_("Data file"),
        upload_to=tools.upload_import_file_to,
        help_text=_("File that contain data to be imported"),
    )
    input_errors_file = models.FileField(
        max_length=512,
        null=True,
        verbose_name=_("Input errors file"),
        help_text=_("File that contain failed rows"),
        upload_to=tools.upload_import_file_to,
    )

    parse_task_id = models.CharField(
        default=str,
        max_length=36,
        verbose_name=_("Parsing task ID"),
        help_text=_("Celery task ID that start `parse_data`"),
    )

    import_task_id = models.CharField(
        max_length=36,
        default=str,
        verbose_name=_("Import task ID"),
        help_text=_("Celery task ID that start `import_data`"),
    )

    parse_finished = models.DateTimeField(
        editable=False,
        null=True,
        verbose_name=_("Parse finished"),
    )

    import_started = models.DateTimeField(
        editable=False,
        null=True,
        verbose_name=_("Import started"),
    )

    import_finished = models.DateTimeField(
        editable=False,
        null=True,
        verbose_name=_("Import finished"),
    )

    skip_parse_step = models.BooleanField(
        default=False,
        help_text=_("Start importing without confirmation"),
        verbose_name=_("Skip parse step"),
    )

    force_import = models.BooleanField(
        default=False,
        help_text=_("Import data with skip invalid rows."),
        verbose_name=_("Force import"),
    )

    class Meta:
        verbose_name = _("Import job")
        verbose_name_plural = _("Import jobs")

    def __str__(self) -> str:
        """Return string representation."""
        resource_name = pathlib.Path(self.resource_path).suffix.lstrip(".")

        return f"ImportJob(resource={resource_name})"

    def save(
        self,
        force_insert: bool = False,
        force_update: bool = False,
        using: str | None = None,
        update_fields: collections.abc.Iterable[str] | None = None,
    ) -> None:
        """Start task for data parsing when ImportJob is created.

        Celery task is manually called with `apply_async`, to provide
        possibility of custom `task_id` with which task will be run.

        """
        is_created = self._state.adding
        self._save_input_errors_file()
        super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )
        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)

    @property
    def resource(self) -> CeleryResource:
        """Get initialized resource instance."""
        resource_class = module_loading.import_string(self.resource_path)
        resource = resource_class(
            **self.resource_kwargs,
        )
        return resource

    @property
    def progress(self) -> TaskStateInfo | None:
        """Return dict with parsing state."""
        current_task_id = (
            self.parse_task_id
            if self.import_status == self.ImportStatus.PARSING
            else self.import_task_id
        )

        if (
            not current_task_id
            or self.import_status not in self.progress_statuses
        ):
            return None

        return self._get_task_state(current_task_id)

    def _check_import_status_correctness(
        self,
        expected_statuses: Sequence[ImportStatus],
    ) -> None:
        """Raise `ValueError` if `ImportJob` is in incorrect state."""
        if self.import_status not in expected_statuses:
            raise ValueError(
                f"ImportJob with id {self.id} has incorrect status: "
                f"`{self.import_status}`. Expected statuses:"
                f" {[status.value for status in expected_statuses]}",
            )

    def start_parse_data_task(self) -> None:
        """Start parsing task."""
        from .. import tasks

        tasks.parse_data_task.apply_async(
            kwargs={"job_id": self.pk},
            task_id=self.parse_task_id,
        )

    def parse_data(self) -> None:
        """Parse `data_file` and collect results.

        Sets `result` and/or `traceback` and update `status`.

        """
        self._check_import_status_correctness(
            expected_statuses=(self.ImportStatus.CREATED,),  # type: ignore[arg-type]
        )

        self.import_status = self.ImportStatus.PARSING
        self.save(update_fields=["import_status"])

        try:
            self.result = self._parse_data_inner()
            self.import_status = (
                self.ImportStatus.INPUT_ERROR
                if self.result.has_errors()
                or self.result.has_validation_errors()
                else self.ImportStatus.PARSED
            )
            self.parse_finished = timezone.now()
            self.save(
                update_fields=[
                    "import_status",
                    "result",
                    "parse_finished",
                ],
            )
        except Exception as error:
            self._handle_error(
                error_message=str(error),
                traceback=traceback.format_exc(),
                exception=error,
            )

    def _parse_data_inner(self) -> Result:
        """Run import process with `dry_run == True`.

        Returns
            apps.utils.async_import_export.results.Result instance with
            parsing results

        """
        dataset = self._get_data_to_import()
        return self.resource.import_data(
            dataset,
            dry_run=True,
            raise_errors=False,
            collect_failed_rows=True,
            force_import=self.force_import,
        )

    def confirm_import(self) -> None:
        """Update task status to IMPORT_CONFIRMED and start parsing.

        This is "intermediate" state between PARSED and IMPORTING and required
        because of possible latency of celery task start.

        Celery task is manually called with `apply_async`, to provide
        possibility of custom `task_id` with which task will be run.

        """
        self._check_import_status_correctness(
            expected_statuses=(self.ImportStatus.PARSED,),  # type: ignore[arg-type]
        )

        self.import_status = self.ImportStatus.CONFIRMED
        self.import_task_id = str(uuid.uuid4())
        self.import_started = timezone.now()
        self.save(
            update_fields=[
                "import_status",
                "import_started",
                "import_task_id",
            ],
        )
        transaction.on_commit(self._start_import_data_task)

    def _start_import_data_task(self) -> None:
        """Start import task."""
        from .. import tasks

        tasks.import_data_task.apply_async(
            kwargs={"job_id": self.pk},
            task_id=self.import_task_id,
        )

    def import_data(self) -> None:
        """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=(expected_status,),  # type: ignore[arg-type]
        )

        self.import_status = self.ImportStatus.IMPORTING
        self.save(update_fields=["import_status"])
        try:
            self.result = self._import_data_inner()
            self.import_status = self.ImportStatus.IMPORTED
            self.import_finished = timezone.now()
            self.save(
                update_fields=[
                    "import_status",
                    "result",
                    "import_finished",
                ],
            )
        except Exception as error:
            self._handle_error(
                error_message=str(error),
                traceback=traceback.format_exc(),
                exception=error,
            )

    def _import_data_inner(self) -> Result:
        """Run import process with saving to DB.

        Transaction is not used as import is slow, so before it finish -
        no instances are saved to DB. So sync works incorrect

        Returns
            import_export.results.Result instance with parsing results

        """
        data_to_import = self._get_data_to_import()
        return self.resource.import_data(
            data_to_import,
            dry_run=False,
            raise_errors=True,
            use_transactions=True,
            collect_failed_rows=True,
            force_import=self.force_import,
        )

    def _get_import_format_by_ext(
        self,
        file_ext: str,
    ) -> type[base_formats.Format]:
        """Determine import file format by file extension."""
        supported_formats = self.resource.get_supported_formats()

        for import_format in supported_formats:
            if import_format().get_title().upper() == file_ext.upper().replace(
                ".",
                "",
            ):
                return import_format

        supported_formats_titles = ",".join(
            supported_format().get_title()
            for supported_format in supported_formats
        )
        raise ValueError(
            f"Incorrect import format: {file_ext}. "
            f"Supported formats: {supported_formats_titles}",
        )

    def _get_data_to_import(self) -> tablib.Dataset:
        """Read ``self.data_file`` content and convert it to dataset."""
        file_ext = pathlib.Path(self.data_file.name).suffix
        input_format = self._get_import_format_by_ext(
            file_ext=file_ext,
        )()
        data = self.data_file.read()
        self.data_file.seek(0)
        if not input_format.is_binary():
            data = encoding.force_str(data)
        data_to_import = input_format.create_dataset(data)
        num_rows = len(data_to_import)
        if num_rows > settings.IMPORT_EXPORT_MAX_DATASET_ROWS:
            raise ValueError(
                f"Too many rows `{num_rows}`"
                f"(Max: {settings.IMPORT_EXPORT_MAX_DATASET_ROWS}). "
                f"Input file may be broken. "
                f"If it's spreadsheet file, please delete empty rows.",
            )
        return data_to_import

    def cancel_import(self) -> None:
        """Cancel current data import.

        ImportJob can be CANCELLED only from following states:
            - CREATED
            - PARSING
            - CONFIRMED
            - IMPORTING

        """
        status_task_field_map = {
            self.ImportStatus.CREATED: "parse_task_id",
            self.ImportStatus.PARSING: "parse_task_id",
            self.ImportStatus.CONFIRMED: "import_task_id",
            self.ImportStatus.IMPORTING: "import_task_id",
        }
        self._check_import_status_correctness(
            expected_statuses=status_task_field_map.keys(),  # type: ignore
        )

        # send signal to celery to revoke task
        task_id = getattr(self, status_task_field_map[self.import_status])
        current_app.control.revoke(task_id, terminate=True)

        self.import_status = self.ImportStatus.CANCELLED
        self.save(update_fields=["import_status"])

    def _get_task_state(self, task_id: str) -> TaskStateInfo:
        """Get state info for passed task_id.

        This method may change job status if task failed, but we did not
        save info about this to DB.

        This may happen if task start import/parsing, but processes was killed.
        In that case, job have status `importing`, but it can't be finished.

        """
        async_result = result.AsyncResult(task_id)
        if async_result.state in states.EXCEPTION_STATES:
            self._handle_error(
                error_message=str(async_result.info),
                traceback=str(async_result.traceback),
            )
            return {
                "state": async_result.state,
                "info": {},
            }
        return {
            "state": async_result.state,
            "info": async_result.info,
        }

    def _handle_error(
        self,
        error_message: str,
        traceback: str,
        exception: Exception | None = None,
    ) -> None:
        """Update job's status in case of error."""
        self.import_status = (
            self.ImportStatus.PARSE_ERROR
            if self.import_status == self.ImportStatus.PARSING
            else self.ImportStatus.IMPORT_ERROR
        )
        error_message_limit = self._meta.get_field("error_message").max_length
        self.error_message = error_message[:error_message_limit]
        self.traceback = traceback
        self.save(
            update_fields=[
                "error_message",
                "traceback",
                "import_status",
            ],
        )
        signals.import_job_failed.send(
            sender=self.__class__,
            instance=self,
            error_message=self.error_message,
            traceback=self.traceback,
            exception=exception,
        )

    def _save_input_errors_file(self) -> None:
        """Save input errors file.

        This should be saved after parsing and after importing
        and if there are row errors in result.

        """
        if (
            self.import_status not in self.results_statuses
            or not self.result
            or not self.result.row_errors()
            or self.input_errors_file
        ):
            return
        file_ext = pathlib.Path(self.data_file.name).suffix
        file_format = self._get_import_format_by_ext(
            file_ext=file_ext,
        )()
        export_data = file_format.export_data(
            dataset=self.result.failed_dataset,
        )

        # create file if `export_data` is not file
        if not hasattr(export_data, "read"):
            export_data = django_files.ContentFile(
                encoding.force_bytes(export_data),
            )

        file_name = self.resource.generate_export_filename(
            file_format,
        ).replace("/", "-")

        self.input_errors_file.save(
            name=f"FailedRows{file_name}",
            content=export_data,
            save=True,
        )

progress property

Return dict with parsing state.

resource property

Get initialized resource instance.

ImportStatus

Bases: TextChoices

ImportJob possible statuses.

• CREATED: import job is just created, no parsing done
• PARSING: parse job started
• PARSED: data_file parsed, no errors in data occurred
• INPUT_ERROR: data_file parsed, data contains errors
• PARSE_ERROR: data_file can't be parsed (invalid format, etc.)
• IMPORT_CONFIRMED import confirmed but not started yet
• IMPORTING: importing job started
• IMPORTED: data from data_file imported to DB w/o errors
• IMPORT_ERROR: unknown error during import
• CANCELLED: import job has been cancelled (revoked)

State diagrams:

CREATED
   |
.parse_data()
   |
PARSING  - (INPUT_ERROR, PARSE_ERROR)
   |
PARSED
   |
.confirm_import()
   |
IMPORT_CONFIRMED
   |
.import_data()
   |
IMPORTING - IMPORT_ERROR
   |
IMPORTED

Source code in import_export_extensions/models/import_job.py

class ImportStatus(models.TextChoices):
    """ImportJob possible statuses.

    * CREATED:
        import job is just created, no parsing done
    * PARSING:
        parse job started
    * PARSED:
        data_file parsed, no errors in data occurred
    * INPUT_ERROR:
        data_file parsed, data contains errors
    * PARSE_ERROR:
        data_file can't be parsed (invalid format, etc.)
    * IMPORT_CONFIRMED
        import confirmed but not started yet
    * IMPORTING:
        importing job started
    * IMPORTED:
        data from data_file imported to DB w/o errors
    * IMPORT_ERROR:
        unknown error during import
    * CANCELLED:
        import job has been cancelled (revoked)

    State diagrams:

        CREATED
           |
        .parse_data()
           |
        PARSING  - (INPUT_ERROR, PARSE_ERROR)
           |
        PARSED
           |
        .confirm_import()
           |
        IMPORT_CONFIRMED
           |
        .import_data()
           |
        IMPORTING - IMPORT_ERROR
           |
        IMPORTED

    """

    CREATED = "CREATED", _("Created")
    PARSING = "PARSING", _("Parsing")
    PARSED = "PARSED", _("Parsed")
    INPUT_ERROR = "INPUT_ERROR", _("Input data error")
    PARSE_ERROR = "PARSE_ERROR", _("Parse error")
    CONFIRMED = "CONFIRMED", _("Import confirmed")
    IMPORTING = "IMPORTING", _("Importing")
    IMPORTED = "IMPORTED", _("Imported")
    IMPORT_ERROR = "IMPORT_ERROR", _("Import error")
    CANCELLED = "CANCELLED", _("Cancelled")

__str__()

Return string representation.

Source code in import_export_extensions/models/import_job.py

def __str__(self) -> str:
    """Return string representation."""
    resource_name = pathlib.Path(self.resource_path).suffix.lstrip(".")

    return f"ImportJob(resource={resource_name})"

cancel_import()

Cancel current data import.

ImportJob can be CANCELLED only from following states

• CREATED
• PARSING
• CONFIRMED
• IMPORTING

Source code in import_export_extensions/models/import_job.py

def cancel_import(self) -> None:
    """Cancel current data import.

    ImportJob can be CANCELLED only from following states:
        - CREATED
        - PARSING
        - CONFIRMED
        - IMPORTING

    """
    status_task_field_map = {
        self.ImportStatus.CREATED: "parse_task_id",
        self.ImportStatus.PARSING: "parse_task_id",
        self.ImportStatus.CONFIRMED: "import_task_id",
        self.ImportStatus.IMPORTING: "import_task_id",
    }
    self._check_import_status_correctness(
        expected_statuses=status_task_field_map.keys(),  # type: ignore
    )

    # send signal to celery to revoke task
    task_id = getattr(self, status_task_field_map[self.import_status])
    current_app.control.revoke(task_id, terminate=True)

    self.import_status = self.ImportStatus.CANCELLED
    self.save(update_fields=["import_status"])

confirm_import()

Update task status to IMPORT_CONFIRMED and start parsing.

This is "intermediate" state between PARSED and IMPORTING and required because of possible latency of celery task start.

Celery task is manually called with apply_async, to provide possibility of custom task_id with which task will be run.

Source code in import_export_extensions/models/import_job.py

def confirm_import(self) -> None:
    """Update task status to IMPORT_CONFIRMED and start parsing.

    This is "intermediate" state between PARSED and IMPORTING and required
    because of possible latency of celery task start.

    Celery task is manually called with `apply_async`, to provide
    possibility of custom `task_id` with which task will be run.

    """
    self._check_import_status_correctness(
        expected_statuses=(self.ImportStatus.PARSED,),  # type: ignore[arg-type]
    )

    self.import_status = self.ImportStatus.CONFIRMED
    self.import_task_id = str(uuid.uuid4())
    self.import_started = timezone.now()
    self.save(
        update_fields=[
            "import_status",
            "import_started",
            "import_task_id",
        ],
    )
    transaction.on_commit(self._start_import_data_task)

import_data()

Import data from data_file to DB.

Source code in import_export_extensions/models/import_job.py

def import_data(self) -> None:
    """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=(expected_status,),  # type: ignore[arg-type]
    )

    self.import_status = self.ImportStatus.IMPORTING
    self.save(update_fields=["import_status"])
    try:
        self.result = self._import_data_inner()
        self.import_status = self.ImportStatus.IMPORTED
        self.import_finished = timezone.now()
        self.save(
            update_fields=[
                "import_status",
                "result",
                "import_finished",
            ],
        )
    except Exception as error:
        self._handle_error(
            error_message=str(error),
            traceback=traceback.format_exc(),
            exception=error,
        )

parse_data()

Parse data_file and collect results.

Sets result and/or traceback and update status.

Source code in import_export_extensions/models/import_job.py

def parse_data(self) -> None:
    """Parse `data_file` and collect results.

    Sets `result` and/or `traceback` and update `status`.

    """
    self._check_import_status_correctness(
        expected_statuses=(self.ImportStatus.CREATED,),  # type: ignore[arg-type]
    )

    self.import_status = self.ImportStatus.PARSING
    self.save(update_fields=["import_status"])

    try:
        self.result = self._parse_data_inner()
        self.import_status = (
            self.ImportStatus.INPUT_ERROR
            if self.result.has_errors()
            or self.result.has_validation_errors()
            else self.ImportStatus.PARSED
        )
        self.parse_finished = timezone.now()
        self.save(
            update_fields=[
                "import_status",
                "result",
                "parse_finished",
            ],
        )
    except Exception as error:
        self._handle_error(
            error_message=str(error),
            traceback=traceback.format_exc(),
            exception=error,
        )

save(force_insert=False, force_update=False, using=None, update_fields=None)

Start task for data parsing when ImportJob is created.

Celery task is manually called with apply_async, to provide possibility of custom task_id with which task will be run.

Source code in import_export_extensions/models/import_job.py

def save(
    self,
    force_insert: bool = False,
    force_update: bool = False,
    using: str | None = None,
    update_fields: collections.abc.Iterable[str] | None = None,
) -> None:
    """Start task for data parsing when ImportJob is created.

    Celery task is manually called with `apply_async`, to provide
    possibility of custom `task_id` with which task will be run.

    """
    is_created = self._state.adding
    self._save_input_errors_file()
    super().save(
        force_insert=force_insert,
        force_update=force_update,
        using=using,
        update_fields=update_fields,
    )
    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)

start_parse_data_task()

Start parsing task.

Source code in import_export_extensions/models/import_job.py

def start_parse_data_task(self) -> None:
    """Start parsing task."""
    from .. import tasks

    tasks.parse_data_task.apply_async(
        kwargs={"job_id": self.pk},
        task_id=self.parse_task_id,
    )