Coverage for apis_core/relations/models.py: 83%

101 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-28 06:34 +0000

1import functools 

2import logging 

3from typing import Optional 

4 

5from django.contrib.contenttypes.fields import GenericForeignKey 

6from django.contrib.contenttypes.models import ContentType 

7from django.core.exceptions import ValidationError 

8from django.db import models 

9from django.db.models import Case, When 

10from django.db.models.base import ModelBase 

11from django.http.request import HttpRequest 

12from django.utils.translation import gettext_lazy as _ 

13from model_utils.managers import InheritanceManager 

14 

15from apis_core.generic.abc import GenericModel 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class RelationManager(InheritanceManager): 

21 def create_between_instances(self, subj, obj, *args, **kwargs): 

22 subj_object_id = subj.pk 

23 subj_content_type = ContentType.objects.get_for_model(subj) 

24 obj_object_id = obj.pk 

25 obj_content_type = ContentType.objects.get_for_model(obj) 

26 rel = self.create( 

27 subj_object_id=subj_object_id, 

28 subj_content_type=subj_content_type, 

29 obj_object_id=obj_object_id, 

30 obj_content_type=obj_content_type, 

31 ) 

32 logger.debug("Created relation %s between %s and %s", rel.name(), subj, obj) 

33 return rel 

34 

35 def to_content_type_with_targets(self, content_type): 

36 """ 

37 Return the queryset annotated with the target content type 

38 and object id, based on the content_type that is passed. 

39 """ 

40 return self.annotate( 

41 target_content_type=Case( 

42 When(subj_content_type=content_type, then="obj_content_type"), 

43 default="subj_content_type", 

44 ), 

45 target_id=Case( 

46 When(subj_content_type=content_type, then="obj_object_id"), 

47 default="subj_object_id", 

48 ), 

49 ) 

50 

51 

52# This ModelBase is simply there to check if the needed attributes 

53# are set in the Relation child classes. 

54class RelationModelBase(ModelBase): 

55 def __new__(metacls, name, bases, attrs): 

56 if name == "Relation": 

57 return super().__new__(metacls, name, bases, attrs) 

58 else: 

59 new_class = super().__new__(metacls, name, bases, attrs) 

60 if not (new_class._meta.abstract or new_class._meta.proxy): 

61 if not hasattr(new_class, "subj_model"): 

62 raise ValueError( 

63 "%s inherits from Relation and must therefore specify subj_model" 

64 % name 

65 ) 

66 if not hasattr(new_class, "obj_model"): 

67 raise ValueError( 

68 "%s inherits from Relation and must therefore specify obj_model" 

69 % name 

70 ) 

71 

72 # `subj_model` or `obj_model` being a list was supported in an earlier 

73 # version of apis, but it is not anymore 

74 if isinstance(getattr(new_class, "subj_model", None), list): 

75 raise ValueError("%s.subj_model must not be a list" % name) 

76 if isinstance(getattr(new_class, "obj_model", None), list): 

77 raise ValueError("%s.obj_model mut not be a list" % name) 

78 

79 if not new_class._meta.ordering: 

80 logger.warning( 

81 f"{name} inherits from Relation but does not specify 'ordering' in its Meta class. " 

82 "Empty ordering could result in inconsitent results with pagination. " 

83 "Set a ordering or inherit the Meta class from Relation.", 

84 ) 

85 

86 return new_class 

87 

88 

89@functools.cache 

90def get_by_natural_key(natural_key: str): 

91 app_label, name = natural_key.lower().split(".") 

92 return ContentType.objects.get_by_natural_key(app_label, name).model_class() 

93 

94 

95class Relation(GenericModel, models.Model, metaclass=RelationModelBase): 

96 class Config(GenericModel.Config): 

97 overview_section = _("Relations") 

98 

99 subj_content_type = models.ForeignKey( 

100 ContentType, on_delete=models.CASCADE, related_name="relation_subj_set" 

101 ) 

102 subj_object_id = models.PositiveIntegerField(null=True) 

103 subj = GenericForeignKey("subj_content_type", "subj_object_id") 

104 obj_content_type = models.ForeignKey( 

105 ContentType, on_delete=models.CASCADE, related_name="relation_obj_set" 

106 ) 

107 obj_object_id = models.PositiveIntegerField(null=True) 

108 obj = GenericForeignKey("obj_content_type", "obj_object_id") 

109 

110 objects = RelationManager() 

111 

112 class Meta: 

113 indexes = [ 

114 models.Index( 

115 fields=["subj_content_type"], name="relations_r_subj_content_type" 

116 ), 

117 models.Index(fields=["subj_object_id"], name="relations_r_subj_object_id"), 

118 models.Index( 

119 fields=["obj_content_type"], name="relations_r_obj_content_type" 

120 ), 

121 models.Index(fields=["obj_object_id"], name="relations_r_obj_object_id"), 

122 models.Index( 

123 fields=["subj_content_type", "subj_object_id"], 

124 name="relations_r_subj_c_t_o_i", 

125 ), 

126 models.Index( 

127 fields=["obj_content_type", "obj_object_id"], 

128 name="relations_r_obj_c_t_o_i", 

129 ), 

130 ] 

131 

132 def save(self, *args, **kwargs): 

133 subj_model = getattr(self, "subj_model", None) 

134 if subj_model and self.subj_content_type.model_class() is not subj_model: 

135 raise ValidationError(f"{self.subj} is not of type {subj_model}") 

136 obj_model = getattr(self, "obj_model", None) 

137 if obj_model and self.obj_content_type.model_class() is not obj_model: 

138 raise ValidationError(f"{self.obj} is not of type {obj_model}") 

139 super().save(*args, **kwargs) 

140 

141 @property 

142 def subj_to_obj_text(self) -> str: 

143 if hasattr(self, "name"): 

144 return f"{self.subj} {self.name()} {self.obj}" 

145 return f"{self.subj} relation to {self.obj}" 

146 

147 @property 

148 def obj_to_subj_text(self) -> str: 

149 if hasattr(self, "reverse_name"): 

150 return f"{self.obj} {self.reverse_name()} {self.subj}" 

151 return f"{self.obj} relation to {self.subj}" 

152 

153 def __str__(self): 

154 return self.subj_to_obj_text 

155 

156 @classmethod 

157 def subj_model_type(cls): 

158 model = cls.subj_model 

159 return get_by_natural_key(model) if isinstance(model, str) else model 

160 

161 @classmethod 

162 def obj_model_type(cls): 

163 model = cls.obj_model 

164 return get_by_natural_key(model) if isinstance(model, str) else model 

165 

166 @classmethod 

167 def name(cls) -> str: 

168 return cls._meta.verbose_name 

169 

170 @classmethod 

171 def reverse_name(cls) -> str: 

172 return cls._meta.verbose_name + " reverse" 

173 

174 @classmethod 

175 def name_and_reverse_name(cls) -> str: 

176 """ 

177 Return a string with both the name and the reverse name. 

178 

179 If they are identical, return only the name. 

180 """ 

181 if cls.name() != cls.reverse_name(): 

182 return f"{cls.name()} - {cls.reverse_name()}" 

183 return cls.name() 

184 

185 def get_update_success_url(self, request: Optional[HttpRequest] = None): 

186 if request and request.GET.get("redirect", False): 

187 return super().get_update_success_url(request) 

188 return self.get_absolute_url()