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