import datetime
import importlib
from dataclasses import dataclass
from datetime import timedelta
from typing import Type
from uuid import UUID
from django.db import models
from django.templatetags.l10n import localize
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.encodings import SupportedEncodings
from ferrosoft.apps.ferrobase.models import (
UUIDModel,
CommonLimit,
StoredFile,
Directory,
LocalUser,
)
[docs]
class UnknownEnumName(RuntimeError):
pass
[docs]
class ImportStatus(models.TextChoices):
NEW = "N", _("New")
PROCESSING = "P", _("Processing")
FINISHED = "F", _("Finished")
ERRORED = "E", _("Errored")
[docs]
def enumerate_models():
from django.apps import apps
return [
(_format_model_name(model), getattr(model._meta, "verbose_name", None))
for model in apps.get_models()
]
def _format_model_name(model: Type[UUIDModel]) -> str:
return "%s.%s" % (model._meta.app_label, model._meta.model_name)
def _guess_related_field(field: models.Field) -> str:
# noinspection PyProtectedMember
for related_field in field.related_model._meta.fields:
if related_field.name == "name" or related_field.name == "iso_code":
return related_field.name
# Fall back to the field declared in the foreign key (even though that
# is likely a UUID).
related_fields = getattr(field, "related_fields", None)
if isinstance(related_fields, list) and len(related_fields) > 0:
return related_fields[0][1].name
raise ValueError("cannot guess related field")
[docs]
class MappingCollection(UUIDModel):
class Meta:
verbose_name = _("Mapping collection")
name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Collection name"),
)
model_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Model name"),
)
documentation = models.TextField(
default="",
help_text=_(
"Free-form HTML snippet to display in format documentation before fields"
),
verbose_name=_("Documentation"),
)
def __str__(self):
return "%s (%s)" % (self.name, self.model_name)
[docs]
@classmethod
def create_mapping_for_model(
cls, name: str, model: Type[UUIDModel]
) -> "MappingCollection":
collection = cls.objects.create(
name=name,
model_name=_format_model_name(model),
)
for field in model._meta.fields:
field_type = FieldType.guess(field)
related_field = related_model = None
required = field_type != FieldType.PRIMARY_KEY and not field.null
if field_type == FieldType.FOREIGN_KEY:
related_model = _format_model_name(field.related_model)
related_field = _guess_related_field(field)
FieldMapping.objects.create(
collection=collection,
field_type=field_type,
input_field_name=field.name,
output_field_name=field.name,
related_model=related_model,
related_field=related_field,
required=required,
)
return collection
[docs]
def get_absolute_url(self) -> str:
return reverse("dataimport:collection_detail", kwargs={"id": self.pk})
[docs]
def get_model(self) -> Type[UUIDModel]:
from django.apps import apps
return apps.get_model(self.model_name)
[docs]
def get_model_without_app(self):
i = self.model_name.find(".")
if i == -1:
return self.model_name
return self.model_name[i + 1 :]
[docs]
class FieldType(models.TextChoices):
BOOLEAN = "BL", _("Boolean")
DATE = "DA", _("Date")
DATE_TIME = "DT", _("Date/Time")
DECIMAL = "DC", _("Decimal")
FOREIGN_KEY = "FK", _("Foreign Key")
PRIMARY_KEY = "PK", _("Primary Key")
TEXT = "TX", _("Text")
INTEGER = "IT", _("Integer")
[docs]
@classmethod
def guess(cls, field: models.Field):
if isinstance(field, models.BooleanField):
return cls.BOOLEAN
elif isinstance(field, (models.CharField, models.TextField)):
return cls.TEXT
elif isinstance(field, models.DecimalField):
return cls.DECIMAL
elif isinstance(field, models.ForeignKey):
return cls.FOREIGN_KEY
elif isinstance(field, models.UUIDField):
return cls.PRIMARY_KEY
elif isinstance(field, models.DateField):
return cls.DATE
elif isinstance(field, models.DateTimeField):
return cls.DATE_TIME
elif isinstance(field, models.IntegerField):
return cls.INTEGER
else:
raise ValueError(
"Field type not supported: {}".format(type(field).__name__)
)
[docs]
class EnumConverter:
def __init__(self, enum_class):
self._enum_class = enum_class
[docs]
def convert(self, value):
if hasattr(self._enum_class, value):
return self._enum_class[value]
raise UnknownEnumName(
_(
"Unsupported value for %(enum)s: %(name)s. Please check the supported values in the documentation."
)
% {
"enum": self._enum_class.__name__,
"name": value,
}
)
[docs]
class FieldMappingManager(models.Manager):
[docs]
def get_ordered(self):
return self.order_by("scope", "input_field_name")
[docs]
class FieldMapping(UUIDModel):
class Meta:
verbose_name = _("Field mapping")
indexes = [
models.Index(
fields=["scope", "input_field_name"],
)
]
objects = FieldMappingManager()
collection = models.ForeignKey(
MappingCollection,
verbose_name=_("Collection"),
on_delete=models.CASCADE,
related_name="field_mappings",
)
field_type = models.CharField(
max_length=2,
choices=FieldType.choices,
verbose_name=_("Field type"),
)
input_field_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Input field name"),
)
output_field_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Output field name"),
)
related_model = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Referenced model"),
default=None,
null=True,
blank=True,
)
related_field = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Referenced field"),
default=None,
null=True,
blank=True,
)
required = models.BooleanField(
verbose_name=_("Required"),
default=True,
)
enum_class = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Fully qualified enumeration class name"),
default=None,
null=True,
blank=True,
)
scope = models.CharField(
blank=True,
default="",
help_text=_(
"A hint to which model the field applies. Some importers "
"create multiple models per line, and this hint clarifies things."
),
verbose_name=_("Scope"),
)
def __str__(self):
related_info = (
""
if self.related_model is None
else " (FK %s.%s)" % (self.related_model, self.related_field)
)
return (
"%(field_type)s: %(input_field_name)s -> %(output_field_name)s%(related_info)s"
% {
"field_type": FieldType(self.field_type).label,
"input_field_name": self.input_field_name,
"output_field_name": self.output_field_name,
"related_info": related_info,
}
)
[docs]
def describe_foreign_key(self):
if FieldType(self.field_type) != FieldType.FOREIGN_KEY:
return ""
related_model = self.get_related_model()
related_field = related_model._meta.get_field(self.related_field)
return _("Reference to %(model)s by %(relation)s") % {
"model": related_model._meta.verbose_name,
"relation": related_field.verbose_name,
}
[docs]
def get_model_field(self):
from django.apps import apps
model = apps.get_model(self.collection.model_name)
return model._meta.get_field(self.output_field_name)
def _get_enum_class(self):
if self.enum_class is None:
return None
module_name, class_name = self.enum_class.rsplit(".", 1)
mod = importlib.import_module(module_name)
if not hasattr(mod, class_name):
raise RuntimeError("Enum class not found: %s" % self.enum_class)
return getattr(mod, class_name)
[docs]
def get_enum_converter(self) -> EnumConverter | None:
enum_class = self._get_enum_class()
if enum_class is None:
return None
return EnumConverter(enum_class)
[docs]
def enum_choices(self):
enum_class = self._get_enum_class()
if enum_class is None:
return []
return [(choice.name, choice.label) for choice in list(enum_class)]
[docs]
@dataclass
class Processors:
before_job: str | None
after_job: str | None
model: str | None
[docs]
class ImportJobMixin(UUIDModel):
DEFAULT_FIELD_SEPARATOR = ","
DEFAULT_DECIMAL_SEPARATOR = "."
DEFAULT_DATE_FORMAT_CODES = "%Y-%m-%d"
DEFAULT_DATE_TIME_FORMAT_CODES = "%Y-%m-%d %H:%M:%S"
name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Name"),
)
text_encoding = models.CharField(
max_length=int(CommonLimit.TINY_CHAR),
choices=SupportedEncodings.choices,
verbose_name=_("Text encoding"),
# Can be null, in which case auto-detect encoding.
null=True,
blank=True,
)
field_separator = models.CharField(
max_length=1,
verbose_name=_("CSV field separator"),
default=DEFAULT_FIELD_SEPARATOR,
blank=True,
)
decimal_separator = models.CharField(
max_length=1,
verbose_name=_("Decimal separator"),
default=DEFAULT_DECIMAL_SEPARATOR,
blank=True,
)
date_format = models.CharField(
max_length=10,
verbose_name=_("Date format"),
choices=DateFormatType.choices,
default=DateFormatType.ISO_8601,
)
date_format_codes = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Date format codes"),
default=DEFAULT_DATE_FORMAT_CODES,
blank=True,
)
date_time_format_codes = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Date/time format codes"),
default=DEFAULT_DATE_TIME_FORMAT_CODES,
blank=True,
)
collection = models.ForeignKey(
MappingCollection,
verbose_name=_("Mapping collection"),
on_delete=models.CASCADE,
)
processor_before_job = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Processor before job"),
help_text=_("Python function to run before models are imported"),
null=True,
blank=True,
default=None,
)
processor_after_job = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Processor after job"),
help_text=_("Python function to run after models are imported"),
null=True,
blank=True,
default=None,
)
model_processor = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Model processor"),
help_text=_("Python function to run on each model saved to database"),
null=True,
blank=True,
default=None,
)
[docs]
class ImportJobTemplate(ImportJobMixin, UUIDModel):
class Meta:
verbose_name = _("Import job template")
code_name = models.CharField(
max_length=int(CommonLimit.TINY_CHAR),
unique=True,
)
def __str__(self):
return f"{self.name}"
[docs]
class ImportJob(ImportJobMixin, UUIDModel):
EXPIRE_FILE_ON = timedelta(days=1)
class Meta:
verbose_name = _("Import job")
creation_time = models.DateTimeField(
verbose_name=_("Creation date"),
)
status = models.CharField(
max_length=1,
choices=ImportStatus.choices,
verbose_name=_("Status"),
)
import_file = models.ForeignKey(
StoredFile,
verbose_name=_("Import file"),
on_delete=models.CASCADE,
related_name="+",
)
line_count = models.PositiveIntegerField(
verbose_name=_("Line count"),
default=0,
)
def __str__(self):
return _("%(name)s from %(date)s") % {
"name": self.name,
"date": localize(self.creation_time),
}
@property
def format_parameters(self) -> FormatParameters:
return FormatParameters(
field_separator=self.field_separator,
decimal_separator=self.decimal_separator,
date_format=DateFormatType(self.date_format),
date_format_codes=self.date_format_codes,
date_time_format_codes=self.date_time_format_codes,
)
[docs]
@classmethod
def default_file_expiry(cls, creation_time: datetime) -> datetime:
return creation_time + cls.EXPIRE_FILE_ON
[docs]
@classmethod
def create(
cls,
import_file,
import_file_name: str,
text_encoding: str | None,
owner: LocalUser,
collection: MappingCollection,
format_params: FormatParameters,
processors: Processors,
):
creation_time = now_utc()
file = StoredFile.objects.create(
name=import_file_name,
content=import_file,
media_type=StoredFile.guess_media_type(import_file_name),
directory=Settings.get_instance().get_import_file_dir(),
creation_time=creation_time,
owner=owner,
expire_on=ImportJob.default_file_expiry(creation_time),
)
return ImportJob.objects.create(
name=import_file_name,
creation_time=creation_time,
status=ImportStatus.NEW,
text_encoding=text_encoding,
import_file=file,
collection=collection,
field_separator=format_params.field_separator,
decimal_separator=format_params.decimal_separator,
date_format=format_params.date_format,
date_format_codes=format_params.date_format_codes,
date_time_format_codes=format_params.date_time_format_codes,
processor_before_job=processors.before_job,
processor_after_job=processors.after_job,
model_processor=processors.model,
)
[docs]
def get_absolute_url(self):
return reverse("dataimport:job_detail", kwargs={"pk": self.pk})
[docs]
def is_done(self) -> bool:
"""Return True if job is surely done, either finished or errored."""
return self.is_status(ImportStatus.FINISHED) or self.is_status(
ImportStatus.ERRORED
)
[docs]
def is_status(self, status: ImportStatus) -> bool:
return self.status == status.value
[docs]
def logs(self):
return LogEntry.objects.filter(import_job=self).order_by("-log_time")
[docs]
def queue_group_name(self):
return "importjob_%s" % str(self.pk)
[docs]
def set_status(self, status: ImportStatus):
self.status = status
self.save(update_fields=["status"])
[docs]
class LogHighlight(models.TextChoices):
NONE = "", _("None")
PRIMARY = "primary", _("Primary")
SECONDARY = "secondary", _("Secondary")
SUCCESS = "success", _("Success")
WARNING = "warning", _("Warning")
DANGER = "danger", _("Danger")
INFO = "info", _("Info")
[docs]
class LogEntry(UUIDModel):
class Meta:
indexes = [
models.Index(fields=["import_job", "log_time"]),
]
import_job = models.ForeignKey(
ImportJob,
verbose_name=_("Import job"),
on_delete=models.CASCADE,
)
log_time = models.DateTimeField(
verbose_name=_("Creation date"),
)
line_no = models.PositiveIntegerField(
verbose_name=_("Line number"),
)
highlight = models.CharField(
max_length=10,
choices=LogHighlight.choices,
default=LogHighlight.NONE,
verbose_name=_("Highlight"),
)
message = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Message"),
)
context = models.JSONField(
verbose_name=_("Context"),
null=True,
default=None,
blank=True,
)
def __str__(self):
return "[%(time)s] [%(line_no)d] %(highlight)s %(message)s" % {
"time": self.log_time.isoformat(),
"line_no": self.line_no,
"highlight": LogHighlight(self.highlight).label,
"message": self.message,
}
[docs]
@classmethod
def create(
cls,
job: ImportJob | UUID,
line_no: int,
message: str,
highlight: LogHighlight = None,
context: dict | None = None,
):
return cls.objects.create(
import_job_id=job.pk if isinstance(job, ImportJob) else job,
log_time=now_utc(),
line_no=line_no,
message=message,
highlight=highlight or LogHighlight.NONE,
context=context or {},
)
[docs]
class UndefinedSettings(Exception):
pass
[docs]
class Settings(UUIDModel):
class Meta:
verbose_name = _("Data import settings")
import_file_dir = models.ForeignKey(
Directory,
verbose_name=_("Import file directory"),
on_delete=models.SET_NULL,
null=True,
default=None,
related_name="+",
)
[docs]
@classmethod
def get_instance(cls) -> "Settings":
it = cls.objects.first()
if it is None:
return cls.objects.create()
return it
[docs]
@classmethod
def get_import_file_dir(cls) -> Directory:
file_dir = cls.get_instance().import_file_dir
if file_dir is None:
raise UndefinedSettings("Import file directory is undefined")
return file_dir
[docs]
@staticmethod
def get_absolute_url():
return reverse("dataimport:settings_index")