Merge 987478f into 45978fa

HISTORY.rst

@@ -2,6 +2,10 @@
 History
 =======
 
+0.4.2 (2023-10-20)
+------------------
+* Add base model for `ImportJob` and `ExportJob`
+
 0.4.1 (2023-09-25)
 ------------------
 * Remvoe ``escape_output`` due it's deprecation

import_export_extensions/migrations/0004_alter_exportjob_created_by_and_more.py

@@ -0,0 +1,112 @@
+# Generated by Django 4.2.5 on 2023-10-06 11:54
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import picklefield.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("import_export_extensions", "0003_importjob_skip_parse_step"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="exportjob",
+            name="created_by",
+            field=models.ForeignKey(
+                editable=False,
+                help_text="User which started job",
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                to=settings.AUTH_USER_MODEL,
+                verbose_name="Created by",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="exportjob",
+            name="error_message",
+            field=models.CharField(
+                blank=True,
+                default=str,
+                help_text="Python error message in case of import/export error",
+                max_length=128,
+                verbose_name="Error message",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="exportjob",
+            name="resource_path",
+            field=models.CharField(
+                help_text="Dotted path to subclass of `import_export.Resource` that should be used for import",
+                max_length=128,
+                verbose_name="Resource class path",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="exportjob",
+            name="result",
+            field=picklefield.fields.PickledObjectField(
+                default=str,
+                editable=False,
+                help_text="Internal job result object that contain info about job statistics. Pickled Python object",
+                verbose_name="Job result",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="exportjob",
+            name="traceback",
+            field=models.TextField(
+                blank=True,
+                default=str,
+                help_text="Python traceback in case of import/export error",
+                verbose_name="Traceback",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="importjob",
+            name="created_by",
+            field=models.ForeignKey(
+                editable=False,
+                help_text="User which started job",
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                to=settings.AUTH_USER_MODEL,
+                verbose_name="Created by",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="importjob",
+            name="error_message",
+            field=models.CharField(
+                blank=True,
+                default=str,
+                help_text="Python error message in case of import/export error",
+                max_length=128,
+                verbose_name="Error message",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="importjob",
+            name="result",
+            field=picklefield.fields.PickledObjectField(
+                default=str,
+                editable=False,
+                help_text="Internal job result object that contain info about job statistics. Pickled Python object",
+                verbose_name="Job result",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="importjob",
+            name="traceback",
+            field=models.TextField(
+                blank=True,
+                default=str,
+                help_text="Python traceback in case of import/export error",
+                verbose_name="Traceback",
+            ),
+        ),
+    ]

import_export_extensions/models/core.py

@@ -1,8 +1,12 @@
 import typing
 
+from django.conf import settings
 from django.db import models
+from django.utils import module_loading
 from django.utils.translation import gettext_lazy as _
 
+from picklefield.fields import PickledObjectField
+
 
 class CreationDateTimeField(models.DateTimeField):
     """DateTimeField to indicate created datetime.
@@ -53,3 +57,68 @@ class TaskStateInfo(typing.TypedDict):
     """Class representing task state dict."""
     state: str
     info: typing.Optional[dict[str, int]]
+
+
+class BaseJob(TimeStampedModel):
+    """Base model for managing celery jobs."""
+
+    resource_path = models.CharField(
+        max_length=128,
+        verbose_name=_("Resource class path"),
+        help_text=_(
+            "Dotted path to subclass of `import_export.Resource` that "
+            "should be used for import",
+        ),
+    )
+    resource_kwargs = models.JSONField(
+        default=dict,
+        verbose_name=_("Resource kwargs"),
+        help_text=_("Keyword parameters required for resource initialization"),
+    )
+    traceback = models.TextField(
+        blank=True,
+        default=str,
+        verbose_name=_("Traceback"),
+        help_text=_("Python traceback in case of import/export error"),
+    )
+    error_message = models.CharField(
+        max_length=128,
+        blank=True,
+        default=str,
+        verbose_name=_("Error message"),
+        help_text=_("Python error message in case of import/export error"),
+    )
+    created_by = models.ForeignKey(
+        to=settings.AUTH_USER_MODEL,
+        editable=False,
+        null=True,
+        on_delete=models.SET_NULL,
+        verbose_name=_("Created by"),
+        help_text=_("User which started job"),
+    )
+    result = PickledObjectField(
+        default=str,
+        verbose_name=_("Job result"),
+        help_text=_(
+            "Internal job result object that contain "
+            "info about job statistics. Pickled Python object",
+        ),
+    )
+
+    class Meta:
+        abstract = True
+
+    @property
+    def resource(self):
+        """Get initialized resource instance."""
+        resource_class = module_loading.import_string(self.resource_path)
+        resource = resource_class(
+            created_by=self.created_by,
+            **self.resource_kwargs,
+        )
+        return resource
+
+    @property
+    def progress(self) -> typing.Optional[TaskStateInfo]:
+        """Return dict with current job state."""
+        raise NotImplementedError

import_export_extensions/models/export_job.py

@@ -3,21 +3,19 @@
 import typing
 import uuid
 
-from django.conf import settings
 from django.core import files as django_files
 from django.db import models, transaction
 from django.utils import encoding, module_loading, timezone
 from django.utils.translation import gettext_lazy as _
 
 from celery import current_app, result, states
 from import_export.formats import base_formats
-from picklefield import PickledObjectField
 
 from . import tools
-from .core import TaskStateInfo, TimeStampedModel
+from .core import BaseJob, TaskStateInfo
 
 
-class ExportJob(TimeStampedModel):
+class ExportJob(BaseJob):
     """Abstract model for managing celery export jobs.
 
     Encapsulate all logic related to celery export.
@@ -72,15 +70,6 @@ class ExportStatus(models.TextChoices):
         verbose_name=_("Job status"),
     )
 
-    resource_path = models.CharField(
-        max_length=128,
-        verbose_name=_("Resource class path"),
-        help_text=_(
-            "Dotted path to subclass of `import_export.Resource` that "
-            "should be used for export",
-        ),
-    )
-
     file_format_path = models.CharField(
         max_length=128,
         verbose_name=_("Export path to class"),
@@ -96,35 +85,6 @@ class ExportStatus(models.TextChoices):
         help_text=_("File that contain exported data"),
     )
 
-    resource_kwargs = models.JSONField(
-        default=dict,
-        verbose_name=_("Resource kwargs"),
-        help_text=_("Keyword parameters required for resource initialization"),
-    )
-
-    traceback = models.TextField(
-        blank=True,
-        default=str,
-        verbose_name=_("Traceback"),
-        help_text=_("Python traceback in case of export error"),
-    )
-    error_message = models.CharField(
-        max_length=128,
-        blank=True,
-        default=str,
-        verbose_name=_("Error message"),
-        help_text=_("Python error message in case of export error"),
-    )
-
-    result = PickledObjectField(
-        default=str,
-        verbose_name=_("Export result"),
-        help_text=_(
-            "Internal export result object that contain "
-            "info about export statistics. Pickled Python object",
-        ),
-    )
-
     export_task_id = models.CharField(  # noqa: DJ01
         verbose_name=_("Export task ID"),
         max_length=36,
@@ -147,15 +107,6 @@ class ExportStatus(models.TextChoices):
         null=True,
     )
 
-    created_by = models.ForeignKey(
-        to=settings.AUTH_USER_MODEL,
-        editable=False,
-        null=True,
-        on_delete=models.SET_NULL,
-        verbose_name=_("Created by"),
-        help_text=_("User which started export"),
-    )
-
     class Meta:
         verbose_name = _("Export job")
         verbose_name_plural = _("Export jobs")
@@ -192,16 +143,6 @@ def save(
             self.save(update_fields=["export_task_id"])
             transaction.on_commit(self._start_export_data_task)
 
-    @property
-    def resource(self):
-        """Get initialized resource instance."""
-        resource_class = module_loading.import_string(self.resource_path)
-        resource = resource_class(
-            created_by=self.created_by,
-            **self.resource_kwargs,
-        )
-        return resource
-
     @property
     def file_format(self) -> base_formats.Format:
         """Get initialized format instance."""
@@ -252,14 +193,14 @@ def progress(self) -> typing.Optional[TaskStateInfo]:
 
         return self._get_task_state(self.export_task_id)
 
-    def _check_import_status_correctness(
+    def _check_export_status_correctness(
         self,
         expected_statuses: typing.Sequence[ExportStatus],
     ) -> None:
-        """Raise `ValueError` if `ImportJob` is in incorrect state."""
+        """Raise `ValueError` if `ExportJob` is in incorrect state."""
         if self.export_status not in expected_statuses:
             raise ValueError(
-                f"ImportJob with id {self.id} has incorrect status: "
+                f"ExportJob with id {self.id} has incorrect status: "
                 f"`{self.export_status}`. Expected statuses:"
                 f" {[status.value for status in expected_statuses]}",
             )
@@ -313,7 +254,7 @@ def cancel_export(self) -> None:
             - EXPORTING
 
         """
-        self._check_import_status_correctness(
+        self._check_export_status_correctness(
             expected_statuses=[
                 self.ExportStatus.CREATED.value,
                 self.ExportStatus.EXPORTING.value,
@@ -367,7 +308,7 @@ def _get_task_state(self, task_id: str) -> TaskStateInfo:
             )
 
         # Update job's status in case of exception
-        self.export_status = ExportJob.ExportStatus.EXPORT_ERROR
+        self.export_status = self.ExportStatus.EXPORT_ERROR
         self.error_message = str(async_result.info)[:128]
         self.traceback = str(async_result.traceback)
         self.save(