Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from django.core.exceptions import FieldDoesNotExist
- from django_extensions.db.fields import json
- from rest_framework.decorators import list_route, detail_route
- from reversion.models import Version
- from common.pagination import SmallPagination, HistoryDiffPagination
- from history.serializers import HistoryUserSerializer
- from history.views import get_serialized_history
- class HistoryDiffPaginator(Paginator):
- @cached_property
- def count(self):
- versions_count = super().count
- if not versions_count:
- return 0
- return versions_count - 1
- def page(self, number):
- """
- The same as Paginator but does top + 1 to get paginated diff list.
- To build one object in diff list we need 2 Versions.
- So for example to get 10 items per page we need operate with 11 Versions.
- """
- number = self.validate_number(number)
- bottom = (number - 1) * self.per_page
- top = bottom + self.per_page
- if top + self.orphans >= self.count:
- top = self.count
- return self._get_page(self.object_list[bottom:top+1], number, self)
- def pk_to_str(model, pk):
- instance = model.objects.filter(pk=pk).first()
- if not instance:
- return pk
- if hasattr(instance, 'name'):
- return instance.name
- return str(instance)
- def get_diff_list(versions_list, instance) -> list:
- """
- Versions MUST be ordered: order_by('revision__date_created')!
- """
- model = instance._meta.model
- diff_list = []
- for h in range(len(versions_list) - 1):
- prev_version = versions_list[h]
- prev_version_data = json.loads(prev_version.serialized_data)[0]['fields']
- new_version = versions_list[h + 1]
- new_version_data = json.loads(new_version.serialized_data)[0]['fields']
- diff_data = []
- for field_name in new_version_data.keys():
- if field_name in ('updated_at', 'updated_by'):
- continue
- new_value = new_version_data.get(field_name)
- prev_value = prev_version_data.get(field_name)
- if new_value != prev_value:
- try:
- model_field = model._meta.get_field(field_name)
- except FieldDoesNotExist: # related DB column was removed
- diff_data.append({
- 'field': field_name,
- 'field_verbose_name': field_name,
- 'previous': prev_value,
- 'new': new_value
- })
- else:
- related_field_type = model_field.get_internal_type()
- if related_field_type == 'ForeignKey':
- related_model = model_field.rel.to
- if new_value:
- new_value = pk_to_str(related_model, new_value)
- if prev_value:
- prev_value = pk_to_str(related_model, prev_value)
- elif related_field_type == 'ManyToManyField':
- related_model = model_field.rel.to
- new_value = ', '.join(
- [pk_to_str(related_model, new_id) for new_id in new_value])
- prev_value = ', '.join(
- [pk_to_str(related_model, prev_id) for prev_id in prev_value])
- diff_data.append({
- 'field': field_name,
- 'field_verbose_name': model_field.verbose_name,
- 'previous': prev_value,
- 'new': new_value
- })
- diff_list.append({
- 'revision_user': HistoryUserSerializer(new_version.revision.user).data,
- 'previous_version': prev_version.pk,
- 'new_version': new_version.pk,
- 'date_created': new_version.revision.date_created,
- 'diff_data': diff_data,
- })
- return diff_list
- class HistoryMixin:
- @detail_route(methods=['get'])
- def object_history(self, request, pk=None, **kwargs):
- instance = self.get_object()
- history = Version.objects.get_for_object(instance).prefetch_related(
- 'object').select_related().select_related('revision__user')
- paginator = SmallPagination()
- result_page = paginator.paginate_queryset(get_serialized_history(history), request)
- return paginator.get_paginated_response(result_page)
- @list_route(methods=['get'])
- def history(self, request, **kwargs):
- model = self.queryset.model
- history = Version.objects.get_for_model(model)
- serialized_history = get_serialized_history(history)
- delete_history = Version.objects.get_deleted(model)
- serialized_history.extend(get_serialized_history(delete_history))
- paginator = SmallPagination()
- result_page = paginator.paginate_queryset(serialized_history, request)
- return paginator.get_paginated_response(result_page)
- @detail_route(methods=['get'])
- def diff_list(self, request, pk=None):
- instance = self.get_object()
- history = Version.objects.get_for_object(instance).prefetch_related(
- 'object').select_related('revision__user').order_by('revision__date_created')
- paginator = HistoryDiffPagination()
- page = paginator.paginate_queryset(history, request)
- return paginator.get_paginated_response(get_diff_list(page, instance=instance))
Add Comment
Please, Sign In to add comment