Coverage for apis_core/relations/signals.py: 53%

1import logging 

2 

3from django.contrib.contenttypes.models import ContentType 

4from django.dispatch import receiver 

5 

6from apis_core.apis_entities.signals import post_merge_with 

7from apis_core.apis_metainfo.signals import post_duplicate 

8from apis_core.relations.models import Relation 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13@receiver(post_duplicate) 

14def copy_relations(sender, instance, duplicate, **kwargs): 

15 logger.info(f"Copying relations from {instance!r} to {duplicate!r}") 

16 content_type = ContentType.objects.get_for_model(instance) 

17 subj_rels = Relation.objects.filter( 

18 subj_content_type=content_type, subj_object_id=instance.id 

19 ).select_subclasses() 

20 obj_rels = Relation.objects.filter( 

21 obj_content_type=content_type, obj_object_id=instance.id 

22 ).select_subclasses() 

23 for rel in subj_rels: 

24 rel.pk = None 

25 rel.id = None 

26 rel.subj_object_id = duplicate.id 

27 rel.save() 

28 for rel in obj_rels: 

29 rel.pk = None 

30 rel.id = None 

31 rel.obj_object_id = duplicate.id 

32 rel.save() 

33 

34 

35@receiver(post_merge_with) 

36def merge_relations(sender, instance, entities, **kwargs): 

37 for ent in entities: 

38 logger.info(f"Merging relations from {ent!r} into {instance!r}") 

39 content_type = ContentType.objects.get_for_model(ent) 

40 Relation.objects.filter( 

41 subj_content_type=content_type, subj_object_id=ent.id 

42 ).update(subj_object_id=instance.id) 

43 Relation.objects.filter( 

44 obj_content_type=content_type, obj_object_id=ent.id 

45 ).update(obj_object_id=instance.id)