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

1import logging 

2 

3from django.contrib.contenttypes.models import ContentType 

4from django.db.models.signals import post_delete, post_save 

5from django.dispatch import receiver 

6 

7from apis_core.generic.signals import post_duplicate, post_merge_with 

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) 

46 

47 

48@receiver(post_delete) 

49def set_relations_null(sender, instance, using, origin, **kwargs): 

50 content_type = ContentType.objects.get_for_model(instance) 

51 object_id = instance.pk 

52 if isinstance(object_id, int): 

53 Relation.objects.filter( 

54 subj_content_type=content_type, subj_object_id=object_id 

55 ).update(subj_object_id=None) 

56 Relation.objects.filter( 

57 obj_content_type=content_type, obj_object_id=object_id 

58 ).update(obj_object_id=None) 

59 

60 

61@receiver(post_save) 

62def create_relations(sender, instance, created, raw, using, update_fields, **kwargs): 

63 """ 

64 This signal looks at the `create_relations_to_uris` attribute of a model 

65 instance. The attribute should contain a dict mapping between a relation 

66 name and mapping between they key `obj` or `subj` and a list of URIs. 

67 The signal then tries to create relations between the instance and the 

68 subjects or objects listed in the relation dict mapping. 

69 An example for the dict mapping would be: 

70 "myapp.livesin" = { 

71 curies = ["https://example.org/123"], 

72 obj = "apis_ontology.place" 

73 } 

74 """ 

75 

76 # disable the handler during fixture loading 

77 if raw: 

78 return 

79 relations = getattr(instance, "create_relations_to_uris", {}) 

80 for relation, details in relations.items(): 

81 relation_model = ContentType.objects.get_by_natural_key( 

82 *relation.split(".") 

83 ).model_class() 

84 

85 target = details.get("obj", None) or details.get("subj", None) 

86 target_content_type = ContentType.objects.get_by_natural_key(*target.split(".")) 

87 related_model = target_content_type.model_class() 

88 

89 for related_uri in details["curies"]: 

90 try: 

91 related_instance = related_model.import_from(uri=related_uri) 

92 if details.get("obj"): 

93 relation_model.object.create_between_instances( 

94 instance, related_instance 

95 ) 

96 else: 

97 relation_model.object.create_between_instances( 

98 related_instance, instance 

99 ) 

100 except Exception as e: 

101 logger.error( 

102 "Could not create relation to %s due to %s", related_uri, e 

103 )