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()