Coverage for apis_core/relations/models.py: 83%
92 statements
« prev ^ index » next coverage.py v7.5.3, created at 2026-01-07 08:21 +0000
« prev ^ index » next coverage.py v7.5.3, created at 2026-01-07 08:21 +0000
1import functools
2import logging
4from django.contrib.contenttypes.fields import GenericForeignKey
5from django.contrib.contenttypes.models import ContentType
6from django.core.exceptions import ValidationError
7from django.db import models
8from django.db.models import Case, When
9from django.db.models.base import ModelBase
10from model_utils.managers import InheritanceManager
12from apis_core.generic.abc import GenericModel
14logger = logging.getLogger(__name__)
17class RelationManager(InheritanceManager):
18 def create_between_instances(self, subj, obj, *args, **kwargs):
19 subj_object_id = subj.pk
20 subj_content_type = ContentType.objects.get_for_model(subj)
21 obj_object_id = obj.pk
22 obj_content_type = ContentType.objects.get_for_model(obj)
23 rel = self.create(
24 subj_object_id=subj_object_id,
25 subj_content_type=subj_content_type,
26 obj_object_id=obj_object_id,
27 obj_content_type=obj_content_type,
28 )
29 logger.debug("Created relation %s between %s and %s", rel.name(), subj, obj)
30 return rel
32 def to_content_type_with_targets(self, content_type):
33 """
34 Return the queryset annotated with the target content type
35 and object id, based on the content_type that is passed.
36 """
37 return self.annotate(
38 target_content_type=Case(
39 When(subj_content_type=content_type, then="obj_content_type"),
40 default="subj_content_type",
41 ),
42 target_id=Case(
43 When(subj_content_type=content_type, then="obj_object_id"),
44 default="subj_object_id",
45 ),
46 )
49# This ModelBase is simply there to check if the needed attributes
50# are set in the Relation child classes.
51class RelationModelBase(ModelBase):
52 def __new__(metacls, name, bases, attrs):
53 if name == "Relation":
54 return super().__new__(metacls, name, bases, attrs)
55 else:
56 new_class = super().__new__(metacls, name, bases, attrs)
57 if not (new_class._meta.abstract or new_class._meta.proxy):
58 if not hasattr(new_class, "subj_model"):
59 raise ValueError(
60 "%s inherits from Relation and must therefore specify subj_model"
61 % name
62 )
63 if not hasattr(new_class, "obj_model"):
64 raise ValueError(
65 "%s inherits from Relation and must therefore specify obj_model"
66 % name
67 )
69 # `subj_model` or `obj_model` being a list was supported in an earlier
70 # version of apis, but it is not anymore
71 if isinstance(getattr(new_class, "subj_model", None), list):
72 raise ValueError("%s.subj_model must not be a list" % name)
73 if isinstance(getattr(new_class, "obj_model", None), list):
74 raise ValueError("%s.obj_model mut not be a list" % name)
76 if not new_class._meta.ordering:
77 logger.warning(
78 f"{name} inherits from Relation but does not specify 'ordering' in its Meta class. "
79 "Empty ordering could result in inconsitent results with pagination. "
80 "Set a ordering or inherit the Meta class from Relation.",
81 )
83 return new_class
86@functools.cache
87def get_by_natural_key(natural_key: str):
88 app_label, name = natural_key.lower().split(".")
89 return ContentType.objects.get_by_natural_key(app_label, name).model_class()
92class Relation(GenericModel, models.Model, metaclass=RelationModelBase):
93 subj_content_type = models.ForeignKey(
94 ContentType, on_delete=models.CASCADE, related_name="relation_subj_set"
95 )
96 subj_object_id = models.PositiveIntegerField(null=True)
97 subj = GenericForeignKey("subj_content_type", "subj_object_id")
98 obj_content_type = models.ForeignKey(
99 ContentType, on_delete=models.CASCADE, related_name="relation_obj_set"
100 )
101 obj_object_id = models.PositiveIntegerField(null=True)
102 obj = GenericForeignKey("obj_content_type", "obj_object_id")
104 objects = RelationManager()
106 class Meta:
107 indexes = [
108 models.Index(
109 fields=["subj_content_type"], name="relations_r_subj_content_type"
110 ),
111 models.Index(fields=["subj_object_id"], name="relations_r_subj_object_id"),
112 models.Index(
113 fields=["obj_content_type"], name="relations_r_obj_content_type"
114 ),
115 models.Index(fields=["obj_object_id"], name="relations_r_obj_object_id"),
116 models.Index(
117 fields=["subj_content_type", "subj_object_id"],
118 name="relations_r_subj_c_t_o_i",
119 ),
120 models.Index(
121 fields=["obj_content_type", "obj_object_id"],
122 name="relations_r_obj_c_t_o_i",
123 ),
124 ]
126 def save(self, *args, **kwargs):
127 subj_model = getattr(self, "subj_model", None)
128 if subj_model and self.subj_content_type is subj_model:
129 raise ValidationError(f"{self.subj} is not of type {subj_model}")
130 obj_model = getattr(self, "obj_model", None)
131 if obj_model and self.obj_content_type is obj_model:
132 raise ValidationError(f"{self.obj} is not of type {obj_model}")
133 super().save(*args, **kwargs)
135 @property
136 def subj_to_obj_text(self) -> str:
137 if hasattr(self, "name"):
138 return f"{self.subj} {self.name()} {self.obj}"
139 return f"{self.subj} relation to {self.obj}"
141 @property
142 def obj_to_subj_text(self) -> str:
143 if hasattr(self, "reverse_name"):
144 return f"{self.obj} {self.reverse_name()} {self.subj}"
145 return f"{self.obj} relation to {self.subj}"
147 def __str__(self):
148 return self.subj_to_obj_text
150 @classmethod
151 def subj_model_type(cls):
152 model = cls.subj_model
153 return get_by_natural_key(model) if isinstance(model, str) else model
155 @classmethod
156 def obj_model_type(cls):
157 model = cls.obj_model
158 return get_by_natural_key(model) if isinstance(model, str) else model
160 @classmethod
161 def name(cls) -> str:
162 return cls._meta.verbose_name
164 @classmethod
165 def reverse_name(cls) -> str:
166 return cls._meta.verbose_name + " reverse"
168 @classmethod
169 def name_and_reverse_name(cls) -> str:
170 """
171 Return a string with both the name and the reverse name.
173 If they are identical, return only the name.
174 """
175 if cls.name() != cls.reverse_name():
176 return f"{cls.name()} - {cls.reverse_name()}"
177 return cls.name()