Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/lifecycle.py: 96%

1from __future__ import annotations 

2 

3from copy import copy 

4from dataclasses import dataclass 

5 

6from django.db import models 

7 

8 

9@dataclass 

10class DeletedProperty: 

11 """Property instance retained long enough to emit its post-delete event.""" 

12 

13 instance: models.Model 

14 

15 

16def before_replay_deletions(changes: list[dict]) -> list[DeletedProperty]: 

17 """Run pre-delete hooks after validation but before generic row deletion.""" 

18 from core.auxiliary.models.PropertyInfo import PropertyInfo 

19 from core.auxiliary.models.PropertyValue import PropertyValue 

20 from core.auxiliary.property_events import property_info_deleting 

21 

22 deleted: list[DeletedProperty] = [] 

23 for change in changes: 

24 if change["action"] != "delete": 

25 continue 

26 queryset = change["spec"].model._base_manager.filter(pk=change["pk"]) 

27 if change["model"] == "core_auxiliary.propertyinfo": 

28 queryset = queryset.select_related("set") 

29 elif change["model"] == "core_auxiliary.propertyvalue": 

30 queryset = queryset.select_related("property", "property__set") 

31 instance = queryset.first() 

32 if instance is None: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 continue 

34 if isinstance(instance, PropertyInfo): 

35 property_info_deleting(instance) 

36 # Django clears the primary key on an object passed to delete(). 

37 # Keep an intact instance for post-delete domain reconciliation. 

38 deleted.append(DeletedProperty(copy(instance))) 

39 elif isinstance(instance, PropertyValue): 

40 # Preserve the cached PropertyInfo because it may be deleted later 

41 # in the same aggregate change-set. 

42 deleted.append(DeletedProperty(copy(instance))) 

43 return deleted 

44 

45 

46def after_replay( 

47 changes: list[dict], 

48 deleted_properties: list[DeletedProperty], 

49) -> None: 

50 """Reconcile property-owned derived state once restored rows are complete.""" 

51 from core.auxiliary.models.PropertyInfo import PropertyInfo 

52 from core.auxiliary.models.PropertyValue import PropertyValue 

53 from core.auxiliary.property_events import ( 

54 property_info_deleted, 

55 property_info_saved, 

56 property_value_deleted, 

57 property_value_saved, 

58 ) 

59 

60 for deleted in deleted_properties: 

61 if isinstance(deleted.instance, PropertyInfo): 

62 property_info_deleted(deleted.instance) 

63 elif isinstance(deleted.instance, PropertyValue): 63 ↛ 60line 63 didn't jump to line 60 because the condition on line 63 was always true

64 property_value_deleted(deleted.instance) 

65 

66 changed_info_ids = { 

67 change["pk"] 

68 for change in changes 

69 if change["model"] == "core_auxiliary.propertyinfo" 

70 and change["action"] in {"create", "update"} 

71 } 

72 changed_value_ids = { 

73 change["pk"] 

74 for change in changes 

75 if change["model"] == "core_auxiliary.propertyvalue" 

76 and change["action"] in {"create", "update"} 

77 } 

78 for property_info in PropertyInfo._base_manager.filter(pk__in=changed_info_ids): 

79 property_info_saved(property_info) 

80 for property_value in PropertyValue._base_manager.filter( 

81 pk__in=changed_value_ids 

82 ).select_related("property"): 

83 property_value_saved(property_value)