Coverage for apis_core/generic/abc.py: 61%
258 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-24 09:48 +0000
1import logging
2import re
3from typing import Optional, Tuple
5from django.contrib.contenttypes.models import ContentType
6from django.core.checks import Error
7from django.core.exceptions import ImproperlyConfigured
8from django.db import models
9from django.db.models import BooleanField, CharField, TextField
10from django.db.models.fields.related import ForeignKey, ManyToManyField
11from django.db.models.query import QuerySet
12from django.forms import model_to_dict
13from django.http.request import HttpRequest
14from django.urls import reverse
15from django.utils.encoding import force_str
16from django.utils.translation import gettext_lazy as _
18from apis_core.generic.helpers import mro_paths, permission_fullname
19from apis_core.generic.signals import (
20 post_duplicate,
21 post_merge_with,
22 pre_duplicate,
23 pre_import_from,
24 pre_merge_with,
25)
26from apis_core.generic.utils import get_autocomplete_data_and_normalized_uri
27from apis_core.generic.utils.models import ConfigModel
28from apis_core.utils.settings import apis_base_uri, rdf_namespace_prefix
30logger = logging.getLogger(__name__)
33class GenericModel(models.Model):
34 class Meta:
35 abstract = True
37 class Config:
38 overview_section = _("Generic models")
39 index_for_search = True
41 @classmethod
42 def check(cls, **kwargs):
43 errors = super().check(**kwargs)
44 msgs = ConfigModel.validation_errors_to_error_messages(cls.Config)
45 errors.extend([Error(msg, obj=cls) for msg in msgs])
46 return errors
48 def __repr__(self):
49 if id := getattr(self, "id", None):
50 return super().__repr__() + f" (ID: {id})"
51 return super().__repr__()
53 @property
54 def get_self_content_type(self):
55 return ContentType.objects.get_for_model(self)
57 @classmethod
58 def get_listview_url(cls):
59 return reverse("apis_core:generic:list", args=[cls])
61 @classmethod
62 def get_createview_url(cls):
63 return reverse("apis_core:generic:create", args=[cls])
65 @classmethod
66 def get_importview_url(cls):
67 return reverse("apis_core:generic:import", args=[cls])
69 @classmethod
70 def get_openapi_tags(cls):
71 return [item[-1] for item in mro_paths(cls)]
73 @classmethod
74 def get_namespace_prefix(cls):
75 ct = ContentType.objects.get_for_model(cls)
76 return f"{rdf_namespace_prefix()}-{ct.model}"
78 @classmethod
79 def get_namespace_uri(cls):
80 return apis_base_uri() + cls.get_listview_url()
82 @classmethod
83 def get_rdf_types(cls):
84 return []
86 @classmethod
87 def get_count(cls):
88 return cls.objects.count()
90 def get_edit_url(self):
91 return reverse("apis_core:generic:update", args=[type(self), self.id])
93 def get_duplicate_url(self):
94 return reverse("apis_core:generic:duplicate", args=[type(self), self.id])
96 def get_enrich_url(self):
97 return reverse("apis_core:generic:enrich", args=[type(self), self.id])
99 def get_absolute_url(self):
100 return reverse("apis_core:generic:detail", args=[type(self), self.id])
102 def get_delete_url(self):
103 return reverse("apis_core:generic:delete", args=[type(self), self.id])
105 def get_merge_url(self, other_id):
106 return reverse("apis_core:generic:merge", args=[type(self), self.id, other_id])
108 def get_select_merge_or_enrich_url(self):
109 return reverse(
110 "apis_core:generic:selectmergeorenrich", args=[type(self), self.id]
111 )
113 def get_create_success_url(self, request: Optional[HttpRequest] = None):
114 if request and request.GET.get("redirect", False):
115 return request.GET.get("redirect")
116 return self.get_absolute_url()
118 def get_update_success_url(self, request: Optional[HttpRequest] = None):
119 if request and request.GET.get("redirect", False):
120 return request.GET.get("redirect")
121 return self.get_edit_url()
123 def get_delete_success_url(self, request: Optional[HttpRequest] = None):
124 if request and request.GET.get("redirect", False):
125 return request.GET.get("redirect")
126 return self.get_listview_url()
128 def get_api_detail_endpoint(self):
129 return reverse(
130 "apis_core:generic:genericmodelapi-detail", args=[type(self), self.id]
131 )
133 @classmethod
134 def get_change_permission(self):
135 return permission_fullname("change", self)
137 @classmethod
138 def get_add_permission(self):
139 return permission_fullname("add", self)
141 @classmethod
142 def get_delete_permission(self):
143 return permission_fullname("delete", self)
145 @classmethod
146 def get_view_permission(self):
147 return permission_fullname("view", self)
149 @classmethod
150 def get_verbose_name_plural(cls):
151 return cls._meta.verbose_name_plural
153 @classmethod
154 def get_verbose_name(cls):
155 return cls._meta.verbose_name
157 @classmethod
158 def valid_import_url(cls, uri: str):
159 """
160 Check if an URI is a can be imported.
161 The exact fetching logic for an URI is defined in the
162 `import_definitions` attribute of the class.
163 `import_definitions` has to be a dict, mapping a regex
164 matching the URI to a callable taking the URI as an argument.
165 This method check if there is a callable defined for this URI.
166 """
167 _, uri = get_autocomplete_data_and_normalized_uri(uri)
168 for regex, fn in getattr(cls, "import_definitions", {}).items():
169 if re.match(regex, uri):
170 return fn
171 return False
173 @classmethod
174 def get_data_and_normalized_uri(cls, uri: str) -> Tuple[dict, str]:
175 data, uri = get_autocomplete_data_and_normalized_uri(uri)
176 return data, uri
178 @classmethod
179 def fetch_from(cls, uri: str):
180 """
181 Normalize the URI and extract the autocomplete data.
182 Then try to fetch data from an URI:
183 Check if there is import logic configured for this URI and if
184 so, use that import logic to fetch the data.
185 Finally, combine the fetched data and the autocomplete data.
186 """
187 logger.debug("Fetch from %s", uri)
188 data, nuri = cls.get_data_and_normalized_uri(uri)
189 if fn := cls.valid_import_url(nuri):
190 fetcheddata = fn(nuri) or {}
191 # merge the two dicts
192 ret = fetcheddata | data
193 # combine values that exist in both dicts
194 for key in set(fetcheddata).intersection(data):
195 ret[key] = fetcheddata[key] + data[key]
196 return ret
197 raise ImproperlyConfigured(f"Import not configured for URI {uri}")
199 @classmethod
200 def import_from(cls, uri: str, allow_empty: bool = True):
201 """
202 Fetch data from an URI and create a model instance using
203 that data. If the `allow_empty` argument is set, this also
204 creates a model instance if the data fetched was empty. This
205 might make sense if you still want to create an instance and
206 attach the URI to it.
207 """
208 # we allow other apps to injercept the import
209 # whatever they return will be used instead of
210 # creating a new object
211 _, nuri = cls.get_data_and_normalized_uri(uri)
212 for receiver, response in pre_import_from.send(sender=cls, uri=nuri):
213 if response:
214 return response
215 data = cls.fetch_from(uri) or {}
216 if allow_empty or data:
217 instance = cls()
218 instance.save()
219 instance.import_data(data)
220 return instance
221 raise ValueError(f"Could not fetch data to import from {uri}")
223 def import_from_dict_subset(self, **data):
224 """
225 Import attributes of this instance from data in a dict.
226 We iterate through the individual values of the dict and
227 a) only set them if the instance has an attribute matching
228 the key and b) use the fields `clean` method to check if
229 the value validates. If it does not validate, we return
230 the validation error in the errors dict.
231 """
232 self._import_errors = {}
233 if data:
234 for field in self._meta.fields:
235 if data.get(field.name, False):
236 value = str(data[field.name][0])
237 try:
238 field.clean(value, self)
239 except Exception as e:
240 logger.info(
241 "Could not set %s on %s: %s", field.name, str(self), str(e)
242 )
243 self._import_errors[field.name] = str(e)
244 else:
245 setattr(self, field.name, value)
246 self.save()
248 def import_data(self, data):
249 self.import_from_dict_subset(**data)
251 def get_merge_charfield_value(self, other: CharField, field: CharField):
252 res = getattr(self, str(field.name))
253 if not field.choices:
254 otherres = getattr(other, str(field.name), res)
255 if not res:
256 res = otherres
257 elif otherres and otherres != res:
258 res += f" ({otherres})"
260 return res
262 def get_merge_textfield_value(self, other: TextField, field: TextField):
263 res = getattr(self, str(field.name))
264 if getattr(other, str(field.name)):
265 # if own value is None, fallback to empty string
266 res = res or ""
267 res += "\n" + f"Merged from {other}:\n" + getattr(other, str(field.name))
268 return res
270 def get_merge_booleanfield(self, other: BooleanField, field: BooleanField):
271 return getattr(other, str(field.name))
273 def get_field_value_after_merge(self, other, field):
274 """
275 This method finds the value of a field after merging `other` into `self`.
276 It first tries to find a merge method that is specific to that field
277 (merge_{fieldname}) and then tries to find a method that is specific to
278 the type of the field (merge_{fieldtype})
279 If neither of those exist, it uses the others field value if the field
280 in self is not set, otherwise it keeps the value in self.
281 """
282 fieldtype = field.get_internal_type().lower()
283 # if there is a `get_merge_{fieldname}` method in this model, use that one
284 if callable(getattr(self, f"get_merge_{field.name}_value", None)):
285 return getattr(self, f"get_merge_{field.name}_value")(other)
286 # otherwise we check if there is a method for the field type and use that one
287 elif callable(getattr(self, f"get_merge_{fieldtype}_value", None)):
288 return getattr(self, f"get_merge_{fieldtype}_value")(other, field)
289 else:
290 if not getattr(self, str(field.name)):
291 return getattr(other, str(field.name))
292 return getattr(self, field.name)
294 def merge_fields(self, other):
295 """
296 This method iterates through the model fields and uses the
297 `get_field_value_after_merge` method to copy values from `other` to `self`.
298 It is called by the `merge_with` method.
299 """
300 for field in self._meta.fields:
301 newval = self.get_field_value_after_merge(other, field)
302 if newval != getattr(self, str(field.name)):
303 setattr(self, str(field.name), newval)
304 self.save()
306 def merge_with(self, entities):
307 if self in entities:
308 entities.remove(self)
309 origin = self.__class__
310 pre_merge_with.send(sender=origin, instance=self, entities=entities)
312 e_a = type(self).__name__
313 self_model_class = ContentType.objects.get(model__iexact=e_a).model_class()
314 if isinstance(entities, int):
315 entities = self_model_class.objects.get(pk=entities)
316 if not isinstance(entities, list) and not isinstance(entities, QuerySet):
317 entities = [entities]
318 entities = [
319 self_model_class.objects.get(pk=ent) if isinstance(ent, int) else ent
320 for ent in entities
321 ]
322 for ent in entities:
323 e_b = type(ent).__name__
324 if e_a != e_b:
325 continue
326 for f in ent._meta.local_many_to_many:
327 if not f.name.endswith("_set"):
328 sl = list(getattr(self, f.name).all())
329 for s in getattr(ent, f.name).all():
330 if s not in sl:
331 getattr(self, f.name).add(s)
333 for ent in entities:
334 self.merge_fields(ent)
336 post_merge_with.send(sender=origin, instance=self, entities=entities)
338 for ent in entities:
339 ent.delete()
341 def duplicate(self):
342 origin = self.__class__
343 pre_duplicate.send(sender=origin, instance=self)
344 # usually, copying instances would work like
345 # https://docs.djangoproject.com/en/4.2/topics/db/queries/#copying-model-instances
346 # but we are working with abstract classes,
347 # so we have to do it by hand using model_to_dict:(
348 objdict = model_to_dict(self)
350 # remove unique fields from dict representation
351 unique_fields = [field for field in self._meta.fields if field.unique]
352 for field in unique_fields:
353 logger.info(f"Duplicating {self}: ignoring unique field {field.name}")
354 objdict.pop(field.name, None)
356 # remove related fields from dict representation
357 related_fields = [
358 field for field in self._meta.get_fields() if field.is_relation
359 ]
360 for field in related_fields:
361 objdict.pop(field.name, None)
363 newobj = type(self).objects.create(**objdict)
365 for field in related_fields:
366 # we are not using `isinstance` because we want to
367 # differentiate between different levels of inheritance
368 if type(field) is ForeignKey:
369 setattr(newobj, field.name, getattr(self, field.name))
370 if type(field) is ManyToManyField:
371 objfield = getattr(newobj, field.name)
372 values = getattr(self, field.name).all()
373 objfield.set(values)
375 newobj.save()
376 post_duplicate.send(sender=origin, instance=self, duplicate=newobj)
377 return newobj
379 duplicate.alters_data = True
381 def uri_set(self):
382 ct = ContentType.objects.get_for_model(self)
383 return (
384 ContentType.objects.get(app_label="uris", model="uri")
385 .model_class()
386 .objects.filter(content_type=ct, object_id=self.id)
387 .all()
388 )
390 def uri_set_with_importer(self):
391 return [uri for uri in self.uri_set() if self.valid_import_url(uri.uri)]
394class SimpleLabelModel(GenericModel):
395 label = models.CharField(
396 blank=True, default="", max_length=4096, verbose_name=_("label")
397 )
399 class Meta:
400 abstract = True
401 ordering = ["label"]
403 def __str__(self):
404 return self.label or force_str(_("No label"))
406 @classmethod
407 def create_from_string(cls, string):
408 return cls.objects.create(label=string)