41
app/core/migrations/0015_modelhistory.py
Normal file
41
app/core/migrations/0015_modelhistory.py
Normal file
@ -0,0 +1,41 @@
|
||||
# Generated by Django 5.1.5 on 2025-02-15 12:10
|
||||
|
||||
import access.fields
|
||||
import access.models
|
||||
import django.db.models.deletion
|
||||
import django.utils.timezone
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('access', '0003_alter_team_organization_organizationnotes_teamnotes'),
|
||||
('contenttypes', '0002_remove_content_type_name'),
|
||||
('core', '0014_data_move_notes_to_new_table'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ModelHistory',
|
||||
fields=[
|
||||
('id', models.AutoField(help_text='ID of the item', primary_key=True, serialize=False, unique=True, verbose_name='ID')),
|
||||
('is_global', models.BooleanField(default=False, help_text='Is this a global object?', verbose_name='Global Object')),
|
||||
('before', models.JSONField(blank=True, default=None, help_text='JSON Object before Change', null=True, verbose_name='Before')),
|
||||
('after', models.JSONField(blank=True, default=None, help_text='JSON Object After Change', null=True, verbose_name='After')),
|
||||
('action', models.IntegerField(choices=[(1, 'Create'), (2, 'Update'), (3, 'Delete')], default=None, help_text='History action performed', null=True, verbose_name='Action')),
|
||||
('created', access.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, help_text='Date and time of creation', verbose_name='Created')),
|
||||
('content_type', models.ForeignKey(blank=True, help_text='Model this note is for', on_delete=django.db.models.deletion.CASCADE, to='contenttypes.contenttype', verbose_name='Content Model')),
|
||||
('organization', models.ForeignKey(help_text='Organization this belongs to', on_delete=django.db.models.deletion.CASCADE, related_name='+', to='access.organization', validators=[access.models.TenancyObject.validatate_organization_exists], verbose_name='Organization')),
|
||||
('user', models.ForeignKey(help_text='User whom performed the action this history relates to', null=True, on_delete=django.db.models.deletion.DO_NOTHING, to=settings.AUTH_USER_MODEL, verbose_name='User')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'History',
|
||||
'verbose_name_plural': 'History',
|
||||
'db_table': 'core_model_history',
|
||||
'ordering': ['-created'],
|
||||
},
|
||||
),
|
||||
]
|
@ -1,9 +1,10 @@
|
||||
import json
|
||||
|
||||
from django.contrib.auth.models import ContentType
|
||||
from django.db import models
|
||||
|
||||
from core.middleware.get_request import get_request
|
||||
from core.models.history import History
|
||||
|
||||
|
||||
|
||||
class SaveHistory(models.Model):
|
||||
@ -21,14 +22,29 @@ class SaveHistory(models.Model):
|
||||
def fields(self):
|
||||
return [ f.name for f in self._meta.fields + self._meta.many_to_many ]
|
||||
|
||||
def save_history(self, before: dict, after: dict):
|
||||
""" Save a Models Changes
|
||||
|
||||
def save_history(self, before: dict, after: dict, history_model = None) -> bool:
|
||||
"""Save Model History
|
||||
|
||||
This method must be re-implemented by the model class in question so
|
||||
that the history model can be passed to this function.
|
||||
|
||||
Args:
|
||||
before (dict): model before saving (model.objects.get().__dict__)
|
||||
after (dict): model after saving and refetched from DB (model.objects.get().__dict__)
|
||||
before (dict): Model data before the change
|
||||
after (dict): Model data after the change
|
||||
history_model(models.Model) History model class
|
||||
|
||||
Returns:
|
||||
False (bool): Failed to save history
|
||||
True (bool): Successfully saved history
|
||||
None (None): history_model was not specified
|
||||
"""
|
||||
|
||||
if history_model is None:
|
||||
|
||||
return None
|
||||
|
||||
|
||||
remove_keys = [
|
||||
'_django_version',
|
||||
'_state',
|
||||
@ -36,6 +52,9 @@ class SaveHistory(models.Model):
|
||||
'modified'
|
||||
]
|
||||
|
||||
|
||||
from core.models.model_history import ModelHistory
|
||||
|
||||
clean = {}
|
||||
for entry in before:
|
||||
|
||||
@ -109,32 +128,23 @@ class SaveHistory(models.Model):
|
||||
|
||||
after_json = json.dumps(clean)
|
||||
|
||||
item_parent_pk = None
|
||||
item_parent_class = None
|
||||
audit_model = self
|
||||
parent_model = None
|
||||
|
||||
if getattr(self, 'parent_object', None):
|
||||
|
||||
parent_model = self.parent_object
|
||||
|
||||
|
||||
if hasattr(self, 'parent_object'):
|
||||
|
||||
if self.parent_object:
|
||||
|
||||
item_parent_pk = self.parent_object.pk
|
||||
item_parent_class = self.parent_object._meta.model_name
|
||||
|
||||
|
||||
item_pk = self.pk
|
||||
action = ModelHistory.Actions.UPDATE
|
||||
|
||||
if not before:
|
||||
|
||||
action = History.Actions.ADD
|
||||
|
||||
elif before_json != after_json and self.pk:
|
||||
|
||||
action = History.Actions.UPDATE
|
||||
action = ModelHistory.Actions.ADD
|
||||
|
||||
elif self.pk is None:
|
||||
|
||||
action = History.Actions.DELETE
|
||||
item_pk = before['id']
|
||||
action = ModelHistory.Actions.DELETE
|
||||
after_json = None
|
||||
|
||||
|
||||
@ -147,22 +157,43 @@ class SaveHistory(models.Model):
|
||||
current_user = None
|
||||
|
||||
|
||||
# if before != after_json and after_json != '{}':
|
||||
if before_json != after_json:
|
||||
entry = History.objects.create(
|
||||
before = before_json,
|
||||
after = after_json,
|
||||
user = current_user,
|
||||
action = action,
|
||||
item_pk = item_pk,
|
||||
item_class = self._meta.model_name,
|
||||
item_parent_pk = item_parent_pk,
|
||||
item_parent_class = item_parent_class,
|
||||
)
|
||||
|
||||
if parent_model is not None:
|
||||
|
||||
entry = history_model.objects.create(
|
||||
organization = self.organization,
|
||||
before = before_json,
|
||||
after = after_json,
|
||||
action = action,
|
||||
user = current_user,
|
||||
content_type = ContentType.objects.get(
|
||||
app_label= self._meta.app_label,
|
||||
model = self._meta.model_name
|
||||
),
|
||||
model = parent_model,
|
||||
child_model = audit_model,
|
||||
)
|
||||
|
||||
else:
|
||||
|
||||
entry = history_model.objects.create(
|
||||
organization = self.organization,
|
||||
before = before_json,
|
||||
after = after_json,
|
||||
action = action,
|
||||
user = current_user,
|
||||
content_type = ContentType.objects.get(
|
||||
app_label= self._meta.app_label,
|
||||
model = self._meta.model_name
|
||||
),
|
||||
model = audit_model,
|
||||
)
|
||||
|
||||
entry.save()
|
||||
|
||||
|
||||
|
||||
def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
|
||||
""" OverRides save for keeping model history.
|
||||
|
||||
@ -187,67 +218,3 @@ class SaveHistory(models.Model):
|
||||
after = self.__dict__.copy()
|
||||
|
||||
self.save_history(before, after)
|
||||
|
||||
|
||||
def delete_history(self, item_pk, item_class):
|
||||
""" Delete the objects history
|
||||
|
||||
When an object is no longer in the database, delete the objects history and
|
||||
that of the child objects. Only caveat is that if the history has a parent_pk
|
||||
the object history is not to be deleted.
|
||||
|
||||
Args:
|
||||
item_pk (int): Primary key of the object to be deleted
|
||||
item_class (str): Object class of the object to be deleted
|
||||
"""
|
||||
|
||||
object_history = History.objects.filter(
|
||||
item_pk = item_pk,
|
||||
item_class = item_class,
|
||||
item_parent_pk = None,
|
||||
)
|
||||
|
||||
if object_history.exists():
|
||||
|
||||
object_history.delete()
|
||||
|
||||
child_object_history = History.objects.filter(
|
||||
item_parent_pk = item_pk,
|
||||
item_parent_class = item_class,
|
||||
)
|
||||
|
||||
if child_object_history.exists():
|
||||
|
||||
child_object_history.delete()
|
||||
|
||||
|
||||
def delete(self, using=None, keep_parents=False):
|
||||
""" OverRides delete for keeping model history and on parent object ONLY!.
|
||||
|
||||
Not a Full-Override as this is just to add to existing.
|
||||
"""
|
||||
|
||||
before = {}
|
||||
item_pk = self.pk
|
||||
item_class = self._meta.model_name
|
||||
|
||||
try:
|
||||
|
||||
before = self.__class__.objects.get(pk=self.pk).__dict__.copy()
|
||||
|
||||
except Exception:
|
||||
|
||||
pass
|
||||
|
||||
# Process the delete
|
||||
super().delete(using=using, keep_parents=keep_parents)
|
||||
|
||||
after = self.__dict__.copy()
|
||||
|
||||
if hasattr(self, 'parent_object'):
|
||||
|
||||
self.save_history(before, after)
|
||||
|
||||
else:
|
||||
|
||||
self.delete_history(item_pk, item_class)
|
||||
|
@ -1,16 +1,20 @@
|
||||
from django.contrib.auth.models import ContentType, User
|
||||
from django.db import models
|
||||
|
||||
from access.fields import *
|
||||
from access.fields import AutoCreatedField
|
||||
from access.models import TenancyObject
|
||||
|
||||
|
||||
|
||||
class ModelHistory(
|
||||
models.Model
|
||||
TenancyObject
|
||||
):
|
||||
|
||||
|
||||
class Meta:
|
||||
|
||||
db_table = 'core_model_history'
|
||||
|
||||
ordering = [
|
||||
'-created'
|
||||
]
|
||||
@ -26,13 +30,7 @@ class ModelHistory(
|
||||
DELETE = 3, 'Delete'
|
||||
|
||||
|
||||
id = models.AutoField(
|
||||
blank=False,
|
||||
help_text = 'ID for this history entry',
|
||||
primary_key=True,
|
||||
unique=True,
|
||||
verbose_name = 'ID'
|
||||
)
|
||||
model_notes = None # model notes not required for this model
|
||||
|
||||
before = models.JSONField(
|
||||
blank = True,
|
||||
@ -83,6 +81,18 @@ class ModelHistory(
|
||||
created = AutoCreatedField()
|
||||
|
||||
|
||||
|
||||
child_history_models = []
|
||||
"""Child History Models
|
||||
|
||||
This list is currently used for excluding child models from the the history
|
||||
select_related query.
|
||||
|
||||
Returns:
|
||||
list: Child history models.
|
||||
"""
|
||||
|
||||
|
||||
table_fields: list = [
|
||||
'created',
|
||||
'action',
|
||||
@ -94,3 +104,29 @@ class ModelHistory(
|
||||
'after'
|
||||
]
|
||||
]
|
||||
|
||||
|
||||
def get_serialized_model_field(self, context):
|
||||
|
||||
model = None
|
||||
|
||||
model = getattr(self, self._meta.related_objects[0].name).model
|
||||
|
||||
model = model.get_serialized_model(context).data
|
||||
|
||||
return model
|
||||
|
||||
|
||||
def get_serialized_child_model_field(self, context):
|
||||
|
||||
model = {}
|
||||
|
||||
parent_model = getattr(self, self._meta.related_objects[0].name)
|
||||
|
||||
child_model = getattr(parent_model, parent_model._meta.related_objects[0].name, None)
|
||||
|
||||
if child_model is not None:
|
||||
|
||||
model = child_model.get_serialized_child_model(context).data
|
||||
|
||||
return model
|
||||
|
@ -1,12 +1,11 @@
|
||||
from django.db.models import Q
|
||||
from django.contrib.auth.models import ContentType
|
||||
|
||||
from drf_spectacular.utils import extend_schema, extend_schema_view, OpenApiResponse
|
||||
|
||||
from api.viewsets.common import ReadOnlyModelViewSet
|
||||
|
||||
from core.serializers.history import (
|
||||
History,
|
||||
HistoryModelSerializer,
|
||||
ModelHistory,
|
||||
HistoryViewSerializer
|
||||
)
|
||||
|
||||
@ -34,13 +33,13 @@ class ViewSet(ReadOnlyModelViewSet):
|
||||
'OPTIONS'
|
||||
]
|
||||
|
||||
|
||||
filterset_fields = [
|
||||
'item_parent_pk',
|
||||
'item_parent_class'
|
||||
'content_type',
|
||||
'user',
|
||||
]
|
||||
|
||||
|
||||
model = History
|
||||
model = ModelHistory
|
||||
|
||||
view_description: str = 'Model Change History'
|
||||
|
||||
@ -51,12 +50,22 @@ class ViewSet(ReadOnlyModelViewSet):
|
||||
|
||||
return self.queryset
|
||||
|
||||
self.queryset = super().get_queryset()
|
||||
history_models = ContentType.objects.filter(
|
||||
model__contains = 'history'
|
||||
).exclude(
|
||||
app_label = 'core',
|
||||
model = 'modelhistory'
|
||||
).exclude(
|
||||
app_label = 'core',
|
||||
model = 'history'
|
||||
).exclude(
|
||||
model__in = self.model.child_history_models
|
||||
)
|
||||
|
||||
self.queryset = self.queryset.filter(
|
||||
Q(item_pk = self.kwargs['model_id'], item_class = self.kwargs['model_class'])
|
||||
|
|
||||
Q(item_parent_pk = self.kwargs['model_id'], item_parent_class = self.kwargs['model_class'])
|
||||
history_models: list = list([ f.model for f in history_models ])
|
||||
|
||||
self.queryset = self.model.objects.select_related(
|
||||
*history_models
|
||||
)
|
||||
|
||||
return self.queryset
|
||||
|
Reference in New Issue
Block a user