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