# (c) cavaliba.com - data - models.py import uuid from django.db import models from django.utils.translation import gettext_lazy as _ # ---------------------------------------------------------------------------------- # DataClass (Schemas) # ---------------------------------------------------------------------------------- class DataClassManager(models.Manager): def get_by_natural_key(self, keyname): return self.get(keyname=keyname) class DataClass(models.Model): keyname = models.SlugField(max_length=128, null=False, blank=False, unique=True) displayname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 is_enabled = models.BooleanField(default=True) page = models.CharField(max_length=255, null=True, blank=True) # noqa: DJ001 order = models.IntegerField(default=100) # V3.19 - YAML (in YAML v3.20: _options: ) options = models.TextField(max_length=5000, null=True, blank=True, default="") # noqa: DJ001 # CRUD permissions per class ; override global built-in ; apply to instances in that class p_create = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_read = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_update = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_delete = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 # no override / all perms p_admin = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 # UI ajax or not is_bigset = models.BooleanField(default=False) # TODO: deprecate ; not a schema related info count_estimation = models.IntegerField(default=0) class Meta: # db_table = "{{ app_name}}_category" db_table = "data_schema" ordering = ["keyname"] verbose_name = "Data Schema" verbose_name_plural = "Data Schemas" def __str__(self): return f"{self.keyname}" def natural_key(self): return self.keyname # ---------------------------------------------------------------------------------- # DataSchema FIELDS # ---------------------------------------------------------------------------------- # default to unknown if not found # *MUST* have : # - a corresponding field_XXXX.py # - a _XXXX.html template # - an entry in data.py FIELD_FORMAT_CHOICE = ( ("string", "string"), ("int", "int"), ("float", "float"), ("boolean", "boolean"), ("ipv4", "ipv4"), ("date", "date"), ("datetime", "datetime"), ("time", "time"), ("schema", "schema"), ("user", "user"), ("group", "group"), ("role", "role"), ("text", "text"), ("enumerate", "enumerate"), ("external", "external"), ("file", "file"), ("password", "password"), # V3.28 ("permission", "permission"), # v4.0 ) class DataSchemaManager(models.Manager): def get_by_natural_key(self, keyname, classname): return self.get(keyname=keyname, classname=classname) class DataSchema(models.Model): classname = models.SlugField("Schema", max_length=128, null=False, blank=False) keyname = models.SlugField(max_length=128, null=False, blank=False) displayname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 description = models.CharField(_("Description"), max_length=500, blank=True) is_enabled = models.BooleanField(default=True) dataformat = models.CharField(choices=FIELD_FORMAT_CHOICE, max_length=32, default="string") dataformat_ext = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 default_value = models.CharField(max_length=500, null=True, blank=True, default="") # noqa: DJ001 page = models.CharField(max_length=255, null=True, blank=True) # noqa: DJ001 order = models.IntegerField(default=100) cardinal_min = models.IntegerField(default=0) cardinal_max = models.IntegerField(default=1) # NEXT - CRUD permissions at field level ###p_create = models.CharField(max_length=256, null=True, blank=True) # p_read = models.CharField(max_length=256, null=True, blank=True) # p_update = models.CharField(max_length=256, null=True, blank=True) ###p_delete = models.CharField(max_length=256, null=True, blank=True) class Meta: db_table = "data_field" unique_together = ["classname", "keyname"] ordering = ["classname", "keyname", "order"] verbose_name = "Data Field" verbose_name_plural = "Data Fields" def __str__(self): return f"{self.classname}:{self.keyname}({self.dataformat}/{self.dataformat_ext})" def natural_key(self): return (self.classname, self.keyname) # ---------------------------------------------------------------------------------- # DataInstance # ---------------------------------------------------------------------------------- class DataInstanceManager(models.Manager): def get_by_natural_key(self, classname, keyname): return self.get(classname=classname, keyname=keyname) class DataInstance(models.Model): classname = models.SlugField("Schema", max_length=128, null=True, blank=True) # noqa: DJ001 keyname = models.CharField(max_length=256, null=False, blank=False) displayname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 is_enabled = models.BooleanField(default=True) data_json = models.TextField(null=True, blank=True) # noqa: DJ001 # CRUD permissions per instance - override per-clas / built-in global p_read = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_update = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_delete = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 last_update = models.DateTimeField(_("Last update"), auto_now_add=False, blank=True, null=True) class Meta: db_table = "data_instance" indexes = [ models.Index(fields=["keyname"]), models.Index(fields=["classname"]), models.Index(fields=["classname", "keyname"]), models.Index(fields=["is_enabled"]), ] unique_together = ["classname", "keyname"] ordering = ["classname", "keyname"] verbose_name = "Data Instance" verbose_name_plural = "Data Instances" def __str__(self): if self.displayname: return f"{self.displayname}" else: return f"{self.keyname}" def natural_key(self): return (self.classname, self.keyname) # ---------------------------------------------------------------------------------- # DataRegistry - v3.16 # ---------------------------------------------------------------------------------- # Simple Key/Value storage class DataRegistry(models.Model): key = models.CharField(max_length=255, null=False, blank=False, unique=True) value = models.TextField(max_length=50000, null=True, blank=True) # noqa: DJ001 # NEXT: date ; created, updated, expire class Meta: db_table = "data_registry" ordering = ["key"] verbose_name = "Registry entry" verbose_name_plural = "Registry entries" def __str__(self): return f"{self.key}" # ---------------------------------------------------------------------------------- # DataEAV - v3.18 # ---------------------------------------------------------------------------------- class DataEAV(models.Model): # instance id (DB pk of instance) iid = models.BigIntegerField(default=0) # instance: classname = models.CharField(max_length=128, null=False, blank=False) keyname = models.CharField(max_length=128, null=False, blank=False) displayname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 is_enabled = models.BooleanField(default=True) fieldname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 format = models.CharField(max_length=512, null=True, blank=True) # noqa: DJ001 value = models.CharField(max_length=1024, null=True, blank=True) # noqa: DJ001 p_read = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 last_update = models.DateTimeField(auto_now_add=False, blank=True, null=True) class Meta: db_table = "data_cache" verbose_name = "Data Cache" verbose_name_plural = "Data Cache" indexes = [ models.Index(fields=["iid"]), models.Index(fields=["classname"]), models.Index(fields=["keyname"]), models.Index(fields=["fieldname"]), models.Index(fields=["format"]), models.Index(fields=["value"]), models.Index(fields=["last_update"]), models.Index(fields=["classname", "keyname"]), models.Index(fields=["classname", "keyname", "fieldname"]), ] def __str__(self): return f"{self.classname}:{self.keyname}:{self.fieldname}" # ---------------------------------------------------------------------------------- # DataFile - v3.19 # ---------------------------------------------------------------------------------- class DataFileManager(models.Manager): def get_by_natural_key(self, fileid): return self.get(fileid=fileid) class DataFile(models.Model): fileid = models.CharField(max_length=128, null=False, blank=False, unique=True) # real filename : myfile.txt filename = models.CharField(max_length=256, null=False, blank=False) displayname = models.CharField(max_length=500, null=True, blank=True) # noqa: DJ001 # store/path: /a/b/c/d/myfile.txt ; s3://a/b/myfile.txt ; relative to Root FileStore conf filepath = models.CharField(max_length=128, null=True, blank=True) # noqa: DJ001 # future / options # fileopts = models.CharField(max_length=512, null=True, blank=True) # is_versioned = models.BooleanField(default=False) p_read = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 p_write = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 size = models.BigIntegerField(null=True, blank=True) hash = models.CharField(max_length=128, null=True, blank=True) # noqa: DJ001 is_encrypted = models.BooleanField(default=False) is_enabled = models.BooleanField(default=True) not_after = models.DateTimeField(auto_now_add=False, blank=True, null=True) last_update = models.DateTimeField(auto_now_add=True, blank=True, null=True) class Meta: db_table = "data_file" verbose_name = "Data File" verbose_name_plural = "Data Files" indexes = [ models.Index(fields=["is_enabled"]), models.Index(fields=["last_update"]), models.Index(fields=["not_after"]), ] def __str__(self): return f"{self.displayname}" def natural_key(self): return self.fileid # ---------------------------------------------------------------------------------- # DataRevision - v3.23 # ---------------------------------------------------------------------------------- class DataRevision(models.Model): date = models.DateTimeField(auto_now_add=True, blank=False, null=False) classname = models.SlugField("Schema", max_length=128, null=False, blank=False) keyname = models.CharField(max_length=256, null=False, blank=False) username = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 displayname = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 action = models.CharField(max_length=256, null=True, blank=True) # noqa: DJ001 # next: data_diff old/new class Meta: db_table = "data_revision" indexes = [ models.Index(fields=["date"]), models.Index(fields=["classname", "keyname"]), ] verbose_name = "Data Revision" verbose_name_plural = "Data Revisions" def __str__(self): return f"{self.date} {self.classname} {self.keyname} {self.displayname} {self.action}" # ---------------------------------------------------------------------------------- # DataTask - async long-running task tracking # v3.33 # ---------------------------------------------------------------------------------- DATATASK_STATE_CHOICES = [ ("QUEUED", "Queued"), ("RUNNING", "Running"), ("DONE", "Done"), ("FAILED", "Failed"), ("ABORTED", "Aborted"), ] DATATASK_OWNER_TYPE_CHOICES = [ ("user", "User"), ("api", "API"), ("auto", "Auto"), ] class DataTask(models.Model): handle = models.UUIDField(default=uuid.uuid4, unique=True, db_index=True) owner_type = models.CharField( max_length=16, choices=DATATASK_OWNER_TYPE_CHOICES, default="user" ) owner_id = models.CharField(max_length=128, db_index=True) name = models.CharField(max_length=256) singleton = models.CharField(max_length=64, blank=True, null=True, db_index=True) # noqa: DJ001 state = models.CharField( max_length=16, choices=DATATASK_STATE_CHOICES, default="QUEUED", db_index=True ) params = models.JSONField(blank=True, null=True) progress = models.JSONField(blank=True, null=True) output = models.JSONField(blank=True, null=True) attachment = models.JSONField(blank=True, null=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) finished_at = models.DateTimeField(blank=True, null=True) duration = models.FloatField(blank=True, null=True) class Meta: db_table = "data_task" ordering = ["-created_at"] verbose_name = "Data Task" verbose_name_plural = "Data Tasks" def __str__(self): return f"{self.name} [{self.state}] {self.handle}"