Coverage for apis_core/uris/signals.py: 98%

1import logging 

2 

3from django.conf import settings 

4from django.contrib.contenttypes.models import ContentType 

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

6from django.dispatch import receiver 

7 

8from apis_core.generic.signals import post_merge_with 

9from apis_core.uris.models import Uri 

10 

11logger = logging.getLogger(__name__) 

12 

13 

14@receiver(post_delete) 

15def remove_stale_uris(sender, instance, *args, **kwargs): 

16 content_type = ContentType.objects.get_for_model(instance) 

17 if isinstance(instance.pk, int): 

18 uris = Uri.objects.filter(content_type=content_type, object_id=instance.pk) 

19 for uri in uris: 

20 logger.info( 

21 "Deleting uri %s as a result of deleting %s", repr(uri), repr(instance) 

22 ) 

23 uri.delete() 

24 

25 

26@receiver(post_save, dispatch_uid="create_default_uri") 

27def create_default_uri(sender, instance, created, raw, using, update_fields, **kwargs): 

28 # disable the handler during fixture loading 

29 if raw: 

30 return 

31 # The list of Uris that should be created 

32 uris = getattr(instance, "_uris", []) 

33 

34 # If this is a new object (created) we ask the model for its 

35 # default Uri and add that to the list of Uris 

36 skip_default_uri = getattr(instance, "skip_default_uri", False) 

37 create_default_uri = getattr(settings, "CREATE_DEFAULT_URI", True) 

38 if created and create_default_uri and not skip_default_uri: 

39 try: 

40 uris.append(instance.get_default_uri()) 

41 except AttributeError: 

42 pass 

43 

44 # We first check if there are even any Uris to create before we 

45 # lookup the content_type. This is a bit of a workaround, because 

46 # during migration the signal is triggered, but the lookup would 

47 # fail. 

48 if uris: 

49 content_type = ContentType.objects.get_for_model(instance) 

50 for uri in uris: 

51 logger.info( 

52 "Creating uri %s as a result of saving %s", repr(uri), repr(instance) 

53 ) 

54 Uri.objects.get_or_create( 

55 uri=uri, content_type=content_type, object_id=instance.id 

56 ) 

57 

58 

59@receiver(post_merge_with) 

60def merge_uris(sender, instance, entities, *args, **kwargs): 

61 instance_content_type = ContentType.objects.get_for_model(instance) 

62 for entity in entities: 

63 content_type = ContentType.objects.get_for_model(entity) 

64 for uri in Uri.objects.filter(content_type=content_type, object_id=entity.id): 

65 logger.info( 

66 "Updating uri %s to point to %s as a result of merging %s", 

67 repr(uri), 

68 repr(instance), 

69 repr(entity), 

70 ) 

71 uri.content_type = instance_content_type 

72 uri.object_id = instance.id 

73 uri.save()