-
Notifications
You must be signed in to change notification settings - Fork 69
/
exportjob.py
140 lines (116 loc) · 3.9 KB
/
exportjob.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
# Copyright (C) 2019 o.s. Auto*Mat
from django.utils import timezone
import json
from author.decorators import with_author
from django.contrib.contenttypes.models import ContentType
from django.db import models
from django.db import transaction
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.utils.translation import gettext_lazy as _
from ..fields import ImportExportFileField
from ..tasks import run_export_job
from ..utils import get_formats
@with_author
class ExportJob(models.Model):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._content_type = None
file = ImportExportFileField(
verbose_name=_("exported file"),
upload_to="django-import-export-celery-export-jobs",
blank=False,
null=False,
max_length=255,
)
processing_initiated = models.DateTimeField(
verbose_name=_("Have we started processing the file? If so when?"),
null=True,
blank=True,
default=None,
)
job_status = models.CharField(
verbose_name=_("Status of the job"),
max_length=160,
blank=True,
)
format = models.CharField(
verbose_name=_("Format of file to be exported"),
max_length=255,
blank=False,
null=True,
)
app_label = models.CharField(
verbose_name=_("App label of model to export from"),
max_length=160,
)
model = models.CharField(
verbose_name=_("Name of model to export from"),
max_length=160,
)
resource = models.CharField(
verbose_name=_("Resource to use when exporting"),
max_length=255,
default="",
)
queryset = models.TextField(
verbose_name=_("JSON list of pks to export"),
null=False,
)
email_on_completion = models.BooleanField(
verbose_name=_("Send me an email when this export job is complete"),
default=True,
)
site_of_origin = models.TextField(
verbose_name=_("Site of origin"),
max_length=255,
default="",
)
class Meta:
verbose_name = _("Export job")
verbose_name_plural = _("Export jobs")
def get_resource_class(self):
if self.resource:
return (
self.get_content_type()
.model_class()
.export_resource_classes()[self.resource][1]
)
def get_content_type(self):
if not self._content_type:
self._content_type = ContentType.objects.get(
app_label=self.app_label,
model=self.model,
)
return self._content_type
def get_queryset(self):
pks = json.loads(self.queryset)
# If customised queryset for the model exists
# then it'll apply filter on that otherwise it'll
# apply filter directly on the model.
resource_class = self.get_resource_class()
if hasattr(resource_class, "get_export_queryset"):
return resource_class().get_export_queryset().filter(pk__in=pks)
return self.get_content_type().model_class().objects.filter(pk__in=pks)
def get_resource_choices(self):
return [
(k, v[0])
for k, v in self.get_content_type()
.model_class()
.export_resource_classes()
.items()
]
@staticmethod
def get_format_choices():
"""returns choices of available export formats"""
return [
(f.CONTENT_TYPE, f().get_title())
for f in get_formats()
if f().can_export()
]
@receiver(post_save, sender=ExportJob)
def exportjob_post_save(sender, instance, **kwargs):
if instance.resource and not instance.processing_initiated:
instance.processing_initiated = timezone.now()
instance.save()
transaction.on_commit(lambda: run_export_job.delay(instance.pk))