2
0

middleware.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. import random
  2. import threading
  3. import uuid
  4. from copy import deepcopy
  5. from datetime import timedelta
  6. from django.conf import settings
  7. from django.db.models.signals import pre_delete, post_save
  8. from django.utils import timezone
  9. from django_prometheus.models import model_deletes, model_inserts, model_updates
  10. from utilities.querysets import DummyQuerySet
  11. from .constants import *
  12. from .models import ObjectChange
  13. from .signals import purge_changelog
  14. from .webhooks import enqueue_webhooks
  15. _thread_locals = threading.local()
  16. def handle_changed_object(sender, instance, **kwargs):
  17. """
  18. Fires when an object is created or updated.
  19. """
  20. # Queue a copy of the object for processing once the request completes
  21. action = OBJECTCHANGE_ACTION_CREATE if kwargs['created'] else OBJECTCHANGE_ACTION_UPDATE
  22. _thread_locals.changed_objects.append(
  23. (instance, action)
  24. )
  25. def handle_deleted_object(sender, instance, **kwargs):
  26. """
  27. Fires when an object is deleted.
  28. """
  29. # Cache custom fields prior to copying the instance
  30. if hasattr(instance, 'cache_custom_fields'):
  31. instance.cache_custom_fields()
  32. # Create a copy of the object being deleted
  33. copy = deepcopy(instance)
  34. # Preserve tags
  35. if hasattr(instance, 'tags'):
  36. copy.tags = DummyQuerySet(instance.tags.all())
  37. # Queue a copy of the object for processing once the request completes
  38. _thread_locals.changed_objects.append(
  39. (copy, OBJECTCHANGE_ACTION_DELETE)
  40. )
  41. def purge_objectchange_cache(sender, **kwargs):
  42. """
  43. Delete any queued object changes waiting to be written.
  44. """
  45. _thread_locals.changed_objects = []
  46. class ObjectChangeMiddleware(object):
  47. """
  48. This middleware performs three functions in response to an object being created, updated, or deleted:
  49. 1. Create an ObjectChange to reflect the modification to the object in the changelog.
  50. 2. Enqueue any relevant webhooks.
  51. 3. Increment the metric counter for the event type.
  52. The post_save and post_delete signals are employed to catch object modifications, however changes are recorded a bit
  53. differently for each. Objects being saved are cached into thread-local storage for action *after* the response has
  54. completed. This ensures that serialization of the object is performed only after any related objects (e.g. tags)
  55. have been created. Conversely, deletions are acted upon immediately, so that the serialized representation of the
  56. object is recorded before it (and any related objects) are actually deleted from the database.
  57. """
  58. def __init__(self, get_response):
  59. self.get_response = get_response
  60. def __call__(self, request):
  61. # Initialize an empty list to cache objects being saved.
  62. _thread_locals.changed_objects = []
  63. # Assign a random unique ID to the request. This will be used to associate multiple object changes made during
  64. # the same request.
  65. request.id = uuid.uuid4()
  66. # Connect our receivers to the post_save and post_delete signals.
  67. post_save.connect(handle_changed_object, dispatch_uid='handle_changed_object')
  68. pre_delete.connect(handle_deleted_object, dispatch_uid='handle_deleted_object')
  69. # Provide a hook for purging the change cache
  70. purge_changelog.connect(purge_objectchange_cache)
  71. # Process the request
  72. response = self.get_response(request)
  73. # If the change cache is empty, there's nothing more we need to do.
  74. if not _thread_locals.changed_objects:
  75. return response
  76. # Create records for any cached objects that were changed.
  77. for instance, action in _thread_locals.changed_objects:
  78. # Record an ObjectChange if applicable
  79. if hasattr(instance, 'to_objectchange'):
  80. objectchange = instance.to_objectchange(action)
  81. objectchange.user = request.user
  82. objectchange.request_id = request.id
  83. objectchange.save()
  84. # Enqueue webhooks
  85. enqueue_webhooks(instance, request.user, request.id, action)
  86. # Increment metric counters
  87. if action == OBJECTCHANGE_ACTION_CREATE:
  88. model_inserts.labels(instance._meta.model_name).inc()
  89. elif action == OBJECTCHANGE_ACTION_UPDATE:
  90. model_updates.labels(instance._meta.model_name).inc()
  91. elif action == OBJECTCHANGE_ACTION_DELETE:
  92. model_deletes.labels(instance._meta.model_name).inc()
  93. # Housekeeping: 1% chance of clearing out expired ObjectChanges. This applies only to requests which result in
  94. # one or more changes being logged.
  95. if settings.CHANGELOG_RETENTION and random.randint(1, 100) == 1:
  96. cutoff = timezone.now() - timedelta(days=settings.CHANGELOG_RETENTION)
  97. purged_count, _ = ObjectChange.objects.filter(
  98. time__lt=cutoff
  99. ).delete()
  100. return response