96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
|
import logging
|
||
|
|
|
||
|
|
from django.db import models
|
||
|
|
from django.db.models.query import QuerySet
|
||
|
|
from django.forms import model_to_dict
|
||
|
|
|
||
|
|
BOOLEAN_CHOICES = (
|
||
|
|
(False, 'No'),
|
||
|
|
(True, 'Yes'),
|
||
|
|
)
|
||
|
|
BOOLEAN_CHOICES_CHAR = (
|
||
|
|
('false', 'No'),
|
||
|
|
('true', 'Yes'),
|
||
|
|
)
|
||
|
|
JOB_STATUS_CHOICES = (
|
||
|
|
('pending', 'Pending'),
|
||
|
|
('running', 'Running'),
|
||
|
|
('success', 'Success'),
|
||
|
|
('error', 'Error'),
|
||
|
|
)
|
||
|
|
|
||
|
|
class TaskAwareQuerySet(QuerySet):
|
||
|
|
def delete(self, task=None):
|
||
|
|
try:
|
||
|
|
if task:
|
||
|
|
task.add_entry(f'Deleting {self.model.__name__}s via Queryset')
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f'Failed to add task entry for {self.model.__name__}s: {e}')
|
||
|
|
finally:
|
||
|
|
return super().delete()
|
||
|
|
|
||
|
|
class TaskAwareModelMixin(models.Model):
|
||
|
|
class Meta:
|
||
|
|
abstract = True
|
||
|
|
objects = TaskAwareQuerySet.as_manager()
|
||
|
|
|
||
|
|
def delete(self, task=None, using=None, keep_parents=False):
|
||
|
|
try:
|
||
|
|
if task:
|
||
|
|
task.add_entry(f'Deleting {self.__class__.__name__} {self}')
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f'Failed to add task entry for {self.__class__.__name__} {self}: {e}')
|
||
|
|
finally:
|
||
|
|
super().delete(using=using, keep_parents=keep_parents)
|
||
|
|
|
||
|
|
|
||
|
|
class DateAwareMixin(models.Model):
|
||
|
|
class Meta:
|
||
|
|
abstract = True
|
||
|
|
created_at = models.DateTimeField(auto_now_add=True, editable=False)
|
||
|
|
updated_at = models.DateTimeField(auto_now=True, editable=False)
|
||
|
|
|
||
|
|
class BaseModel(DateAwareMixin):
|
||
|
|
_old_values = {}
|
||
|
|
|
||
|
|
class Meta:
|
||
|
|
abstract = True
|
||
|
|
|
||
|
|
def __init__(self, *args, **kwargs):
|
||
|
|
super().__init__(*args, **kwargs)
|
||
|
|
self._old_values = self.to_json
|
||
|
|
|
||
|
|
internal_id = models.BigAutoField(primary_key=True)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def to_json(self):
|
||
|
|
return model_to_dict(self)
|
||
|
|
|
||
|
|
def write(self, throw_on_error=False, **kwargs):
|
||
|
|
logging.debug(f'Writing {self} to DB with {kwargs}')
|
||
|
|
for field, value in kwargs.items():
|
||
|
|
if hasattr(self, field):
|
||
|
|
logging.debug(f'Setting {field} to {value} for {self}')
|
||
|
|
setattr(self, field, value)
|
||
|
|
else:
|
||
|
|
if throw_on_error:
|
||
|
|
raise AttributeError(f'Could not find {field} in {self.__class__.__name__}')
|
||
|
|
logging.warning(f'Could not find {field} in {self.__class__.__name__}')
|
||
|
|
if not self._state.adding:
|
||
|
|
self.save(update_fields=kwargs.keys())
|
||
|
|
else:
|
||
|
|
logging.warning(f'Trying to write {self} to DB - object is not yet saved')
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
class SearchableMixin:
|
||
|
|
@classmethod
|
||
|
|
def term_filter(cls, search_string):
|
||
|
|
raise NotImplemented('"{cls.__class__.__name__}.search_by_term" not implemented')
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def term_search(cls, search_string):
|
||
|
|
if search_string:
|
||
|
|
return cls.objects.filter(cls.term_filter(search_string))
|
||
|
|
return cls.objects.all()
|