utils.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614
  1. import datetime
  2. import decimal
  3. import json
  4. from collections import defaultdict
  5. from functools import reduce
  6. from operator import or_
  7. from django.core.exceptions import FieldDoesNotExist
  8. from django.core.validators import EMPTY_VALUES
  9. from django.db import models, router
  10. from django.db.models.constants import LOOKUP_SEP
  11. from django.db.models.deletion import Collector
  12. from django.forms.utils import pretty_name
  13. from django.urls import NoReverseMatch, reverse
  14. from django.utils import formats, timezone
  15. from django.utils.hashable import make_hashable
  16. from django.utils.html import format_html
  17. from django.utils.regex_helper import _lazy_re_compile
  18. from django.utils.text import capfirst
  19. from django.utils.translation import ngettext
  20. from django.utils.translation import override as translation_override
  21. QUOTE_MAP = {i: "_%02X" % i for i in b'":/_#?;@&=+$,"[]<>%\n\\'}
  22. UNQUOTE_MAP = {v: chr(k) for k, v in QUOTE_MAP.items()}
  23. UNQUOTE_RE = _lazy_re_compile("_(?:%s)" % "|".join([x[1:] for x in UNQUOTE_MAP]))
  24. class FieldIsAForeignKeyColumnName(Exception):
  25. """A field is a foreign key attname, i.e. <FK>_id."""
  26. pass
  27. def lookup_spawns_duplicates(opts, lookup_path):
  28. """
  29. Return True if the given lookup path spawns duplicates.
  30. """
  31. lookup_fields = lookup_path.split(LOOKUP_SEP)
  32. # Go through the fields (following all relations) and look for an m2m.
  33. for field_name in lookup_fields:
  34. if field_name == "pk":
  35. field_name = opts.pk.name
  36. try:
  37. field = opts.get_field(field_name)
  38. except FieldDoesNotExist:
  39. # Ignore query lookups.
  40. continue
  41. else:
  42. if hasattr(field, "path_infos"):
  43. # This field is a relation; update opts to follow the relation.
  44. path_info = field.path_infos
  45. opts = path_info[-1].to_opts
  46. if any(path.m2m for path in path_info):
  47. # This field is a m2m relation so duplicates must be
  48. # handled.
  49. return True
  50. return False
  51. def get_last_value_from_parameters(parameters, key):
  52. value = parameters.get(key)
  53. return value[-1] if isinstance(value, list) else value
  54. def prepare_lookup_value(key, value, separator=","):
  55. """
  56. Return a lookup value prepared to be used in queryset filtering.
  57. """
  58. if isinstance(value, list):
  59. return [prepare_lookup_value(key, v, separator=separator) for v in value]
  60. # if key ends with __in, split parameter into separate values
  61. if key.endswith("__in"):
  62. value = value.split(separator)
  63. # if key ends with __isnull, special case '' and the string literals 'false' and '0'
  64. elif key.endswith("__isnull"):
  65. value = value.lower() not in ("", "false", "0")
  66. return value
  67. def build_q_object_from_lookup_parameters(parameters):
  68. q_object = models.Q()
  69. for param, param_item_list in parameters.items():
  70. q_object &= reduce(or_, (models.Q((param, item)) for item in param_item_list))
  71. return q_object
  72. def quote(s):
  73. """
  74. Ensure that primary key values do not confuse the admin URLs by escaping
  75. any '/', '_' and ':' and similarly problematic characters.
  76. Similar to urllib.parse.quote(), except that the quoting is slightly
  77. different so that it doesn't get automatically unquoted by the web browser.
  78. """
  79. return s.translate(QUOTE_MAP) if isinstance(s, str) else s
  80. def unquote(s):
  81. """Undo the effects of quote()."""
  82. return UNQUOTE_RE.sub(lambda m: UNQUOTE_MAP[m[0]], s)
  83. def flatten(fields):
  84. """
  85. Return a list which is a single level of flattening of the original list.
  86. """
  87. flat = []
  88. for field in fields:
  89. if isinstance(field, (list, tuple)):
  90. flat.extend(field)
  91. else:
  92. flat.append(field)
  93. return flat
  94. def flatten_fieldsets(fieldsets):
  95. """Return a list of field names from an admin fieldsets structure."""
  96. field_names = []
  97. for name, opts in fieldsets:
  98. field_names.extend(flatten(opts["fields"]))
  99. return field_names
  100. def get_deleted_objects(objs, request, admin_site):
  101. """
  102. Find all objects related to ``objs`` that should also be deleted. ``objs``
  103. must be a homogeneous iterable of objects (e.g. a QuerySet).
  104. Return a nested list of strings suitable for display in the
  105. template with the ``unordered_list`` filter.
  106. """
  107. try:
  108. obj = objs[0]
  109. except IndexError:
  110. return [], {}, set(), []
  111. else:
  112. using = router.db_for_write(obj._meta.model)
  113. collector = NestedObjects(using=using, origin=objs)
  114. collector.collect(objs)
  115. perms_needed = set()
  116. def format_callback(obj):
  117. model = obj.__class__
  118. opts = obj._meta
  119. no_edit_link = "%s: %s" % (capfirst(opts.verbose_name), obj)
  120. if admin_site.is_registered(model):
  121. if not admin_site.get_model_admin(model).has_delete_permission(
  122. request, obj
  123. ):
  124. perms_needed.add(opts.verbose_name)
  125. try:
  126. admin_url = reverse(
  127. "%s:%s_%s_change"
  128. % (admin_site.name, opts.app_label, opts.model_name),
  129. None,
  130. (quote(obj.pk),),
  131. )
  132. except NoReverseMatch:
  133. # Change url doesn't exist -- don't display link to edit
  134. return no_edit_link
  135. # Display a link to the admin page.
  136. return format_html(
  137. '{}: <a href="{}">{}</a>', capfirst(opts.verbose_name), admin_url, obj
  138. )
  139. else:
  140. # Don't display link to edit, because it either has no
  141. # admin or is edited inline.
  142. return no_edit_link
  143. to_delete = collector.nested(format_callback)
  144. protected = [format_callback(obj) for obj in collector.protected]
  145. model_count = {
  146. model._meta.verbose_name_plural: len(objs)
  147. for model, objs in collector.model_objs.items()
  148. }
  149. return to_delete, model_count, perms_needed, protected
  150. class NestedObjects(Collector):
  151. def __init__(self, *args, **kwargs):
  152. super().__init__(*args, **kwargs)
  153. self.edges = {} # {from_instance: [to_instances]}
  154. self.protected = set()
  155. self.model_objs = defaultdict(set)
  156. def add_edge(self, source, target):
  157. self.edges.setdefault(source, []).append(target)
  158. def collect(self, objs, source=None, source_attr=None, **kwargs):
  159. for obj in objs:
  160. if source_attr and not source_attr.endswith("+"):
  161. related_name = source_attr % {
  162. "class": source._meta.model_name,
  163. "app_label": source._meta.app_label,
  164. }
  165. self.add_edge(getattr(obj, related_name), obj)
  166. else:
  167. self.add_edge(None, obj)
  168. self.model_objs[obj._meta.model].add(obj)
  169. try:
  170. return super().collect(objs, source_attr=source_attr, **kwargs)
  171. except models.ProtectedError as e:
  172. self.protected.update(e.protected_objects)
  173. except models.RestrictedError as e:
  174. self.protected.update(e.restricted_objects)
  175. def related_objects(self, related_model, related_fields, objs):
  176. qs = super().related_objects(related_model, related_fields, objs)
  177. return qs.select_related(
  178. *[related_field.name for related_field in related_fields]
  179. )
  180. def _nested(self, obj, seen, format_callback):
  181. if obj in seen:
  182. return []
  183. seen.add(obj)
  184. children = []
  185. for child in self.edges.get(obj, ()):
  186. children.extend(self._nested(child, seen, format_callback))
  187. if format_callback:
  188. ret = [format_callback(obj)]
  189. else:
  190. ret = [obj]
  191. if children:
  192. ret.append(children)
  193. return ret
  194. def nested(self, format_callback=None):
  195. """
  196. Return the graph as a nested list.
  197. """
  198. seen = set()
  199. roots = []
  200. for root in self.edges.get(None, ()):
  201. roots.extend(self._nested(root, seen, format_callback))
  202. return roots
  203. def can_fast_delete(self, *args, **kwargs):
  204. """
  205. We always want to load the objects into memory so that we can display
  206. them to the user in confirm page.
  207. """
  208. return False
  209. def model_format_dict(obj):
  210. """
  211. Return a `dict` with keys 'verbose_name' and 'verbose_name_plural',
  212. typically for use with string formatting.
  213. `obj` may be a `Model` instance, `Model` subclass, or `QuerySet` instance.
  214. """
  215. if isinstance(obj, (models.Model, models.base.ModelBase)):
  216. opts = obj._meta
  217. elif isinstance(obj, models.query.QuerySet):
  218. opts = obj.model._meta
  219. else:
  220. opts = obj
  221. return {
  222. "verbose_name": opts.verbose_name,
  223. "verbose_name_plural": opts.verbose_name_plural,
  224. }
  225. def model_ngettext(obj, n=None):
  226. """
  227. Return the appropriate `verbose_name` or `verbose_name_plural` value for
  228. `obj` depending on the count `n`.
  229. `obj` may be a `Model` instance, `Model` subclass, or `QuerySet` instance.
  230. If `obj` is a `QuerySet` instance, `n` is optional and the length of the
  231. `QuerySet` is used.
  232. """
  233. if isinstance(obj, models.query.QuerySet):
  234. if n is None:
  235. n = obj.count()
  236. obj = obj.model
  237. d = model_format_dict(obj)
  238. singular, plural = d["verbose_name"], d["verbose_name_plural"]
  239. return ngettext(singular, plural, n or 0)
  240. def lookup_field(name, obj, model_admin=None):
  241. opts = obj._meta
  242. try:
  243. f = _get_non_gfk_field(opts, name)
  244. except (FieldDoesNotExist, FieldIsAForeignKeyColumnName):
  245. # For non-regular field values, the value is either a method,
  246. # property, related field, or returned via a callable.
  247. if callable(name):
  248. attr = name
  249. value = attr(obj)
  250. elif hasattr(model_admin, name) and name != "__str__":
  251. attr = getattr(model_admin, name)
  252. value = attr(obj)
  253. else:
  254. sentinel = object()
  255. attr = getattr(obj, name, sentinel)
  256. if callable(attr):
  257. value = attr()
  258. else:
  259. if attr is sentinel:
  260. attr = obj
  261. for part in name.split(LOOKUP_SEP):
  262. attr = getattr(attr, part, sentinel)
  263. if attr is sentinel:
  264. return None, None, None
  265. value = attr
  266. if hasattr(model_admin, "model") and hasattr(model_admin.model, name):
  267. attr = getattr(model_admin.model, name)
  268. f = None
  269. else:
  270. attr = None
  271. value = getattr(obj, name)
  272. return f, attr, value
  273. def _get_non_gfk_field(opts, name):
  274. """
  275. For historical reasons, the admin app relies on GenericForeignKeys as being
  276. "not found" by get_field(). This could likely be cleaned up.
  277. Reverse relations should also be excluded as these aren't attributes of the
  278. model (rather something like `foo_set`).
  279. """
  280. field = opts.get_field(name)
  281. if (
  282. field.is_relation
  283. and
  284. # Generic foreign keys OR reverse relations
  285. ((field.many_to_one and not field.related_model) or field.one_to_many)
  286. ):
  287. raise FieldDoesNotExist()
  288. # Avoid coercing <FK>_id fields to FK
  289. if (
  290. field.is_relation
  291. and not field.many_to_many
  292. and hasattr(field, "attname")
  293. and field.attname == name
  294. ):
  295. raise FieldIsAForeignKeyColumnName()
  296. return field
  297. def label_for_field(name, model, model_admin=None, return_attr=False, form=None):
  298. """
  299. Return a sensible label for a field name. The name can be a callable,
  300. property (but not created with @property decorator), or the name of an
  301. object's attribute, as well as a model field, including across related
  302. objects. If return_attr is True, also return the resolved attribute
  303. (which could be a callable). This will be None if (and only if) the name
  304. refers to a field.
  305. """
  306. attr = None
  307. try:
  308. field = _get_non_gfk_field(model._meta, name)
  309. try:
  310. label = field.verbose_name
  311. except AttributeError:
  312. # field is likely a ForeignObjectRel
  313. label = field.related_model._meta.verbose_name
  314. except FieldDoesNotExist:
  315. if name == "__str__":
  316. label = str(model._meta.verbose_name)
  317. attr = str
  318. else:
  319. if callable(name):
  320. attr = name
  321. elif hasattr(model_admin, name):
  322. attr = getattr(model_admin, name)
  323. elif hasattr(model, name):
  324. attr = getattr(model, name)
  325. elif form and name in form.fields:
  326. attr = form.fields[name]
  327. else:
  328. try:
  329. attr = get_fields_from_path(model, name)[-1]
  330. except (FieldDoesNotExist, NotRelationField):
  331. message = f"Unable to lookup '{name}' on {model._meta.object_name}"
  332. if model_admin:
  333. message += f" or {model_admin.__class__.__name__}"
  334. if form:
  335. message += f" or {form.__class__.__name__}"
  336. raise AttributeError(message)
  337. if hasattr(attr, "short_description"):
  338. label = attr.short_description
  339. elif (
  340. isinstance(attr, property)
  341. and hasattr(attr, "fget")
  342. and hasattr(attr.fget, "short_description")
  343. ):
  344. label = attr.fget.short_description
  345. elif callable(attr):
  346. if attr.__name__ == "<lambda>":
  347. label = "--"
  348. else:
  349. label = pretty_name(attr.__name__)
  350. else:
  351. label = pretty_name(name)
  352. except FieldIsAForeignKeyColumnName:
  353. label = pretty_name(name)
  354. attr = name
  355. if return_attr:
  356. return (label, attr)
  357. else:
  358. return label
  359. def help_text_for_field(name, model):
  360. help_text = ""
  361. try:
  362. field = _get_non_gfk_field(model._meta, name)
  363. except (FieldDoesNotExist, FieldIsAForeignKeyColumnName):
  364. pass
  365. else:
  366. if hasattr(field, "help_text"):
  367. help_text = field.help_text
  368. return help_text
  369. def display_for_field(value, field, empty_value_display):
  370. from django.contrib.admin.templatetags.admin_list import _boolean_icon
  371. if getattr(field, "flatchoices", None):
  372. try:
  373. return dict(field.flatchoices).get(value, empty_value_display)
  374. except TypeError:
  375. # Allow list-like choices.
  376. flatchoices = make_hashable(field.flatchoices)
  377. value = make_hashable(value)
  378. return dict(flatchoices).get(value, empty_value_display)
  379. # BooleanField needs special-case null-handling, so it comes before the
  380. # general null test.
  381. elif isinstance(field, models.BooleanField):
  382. return _boolean_icon(value)
  383. elif value in field.empty_values:
  384. return empty_value_display
  385. elif isinstance(field, models.DateTimeField):
  386. return formats.localize(timezone.template_localtime(value))
  387. elif isinstance(field, (models.DateField, models.TimeField)):
  388. return formats.localize(value)
  389. elif isinstance(field, models.DecimalField):
  390. return formats.number_format(value, field.decimal_places)
  391. elif isinstance(field, (models.IntegerField, models.FloatField)):
  392. return formats.number_format(value)
  393. elif isinstance(field, models.FileField) and value:
  394. return format_html('<a href="{}">{}</a>', value.url, value)
  395. elif isinstance(field, models.JSONField) and value:
  396. try:
  397. return json.dumps(value, ensure_ascii=False, cls=field.encoder)
  398. except TypeError:
  399. return display_for_value(value, empty_value_display)
  400. else:
  401. return display_for_value(value, empty_value_display)
  402. def display_for_value(value, empty_value_display, boolean=False):
  403. from django.contrib.admin.templatetags.admin_list import _boolean_icon
  404. if boolean:
  405. return _boolean_icon(value)
  406. elif value in EMPTY_VALUES:
  407. return empty_value_display
  408. elif isinstance(value, bool):
  409. return str(value)
  410. elif isinstance(value, datetime.datetime):
  411. return formats.localize(timezone.template_localtime(value))
  412. elif isinstance(value, (datetime.date, datetime.time)):
  413. return formats.localize(value)
  414. elif isinstance(value, (int, decimal.Decimal, float)):
  415. return formats.number_format(value)
  416. elif isinstance(value, (list, tuple)):
  417. return ", ".join(str(v) for v in value)
  418. else:
  419. return str(value)
  420. class NotRelationField(Exception):
  421. pass
  422. def get_model_from_relation(field):
  423. if hasattr(field, "path_infos"):
  424. return field.path_infos[-1].to_opts.model
  425. else:
  426. raise NotRelationField
  427. def reverse_field_path(model, path):
  428. """Create a reversed field path.
  429. E.g. Given (Order, "user__groups"),
  430. return (Group, "user__order").
  431. Final field must be a related model, not a data field.
  432. """
  433. reversed_path = []
  434. parent = model
  435. pieces = path.split(LOOKUP_SEP)
  436. for piece in pieces:
  437. field = parent._meta.get_field(piece)
  438. # skip trailing data field if extant:
  439. if len(reversed_path) == len(pieces) - 1: # final iteration
  440. try:
  441. get_model_from_relation(field)
  442. except NotRelationField:
  443. break
  444. # Field should point to another model
  445. if field.is_relation and not (field.auto_created and not field.concrete):
  446. related_name = field.related_query_name()
  447. parent = field.remote_field.model
  448. else:
  449. related_name = field.field.name
  450. parent = field.related_model
  451. reversed_path.insert(0, related_name)
  452. return (parent, LOOKUP_SEP.join(reversed_path))
  453. def get_fields_from_path(model, path):
  454. """Return list of Fields given path relative to model.
  455. e.g. (ModelX, "user__groups__name") -> [
  456. <django.db.models.fields.related.ForeignKey object at 0x...>,
  457. <django.db.models.fields.related.ManyToManyField object at 0x...>,
  458. <django.db.models.fields.CharField object at 0x...>,
  459. ]
  460. """
  461. pieces = path.split(LOOKUP_SEP)
  462. fields = []
  463. for piece in pieces:
  464. if fields:
  465. parent = get_model_from_relation(fields[-1])
  466. else:
  467. parent = model
  468. fields.append(parent._meta.get_field(piece))
  469. return fields
  470. def construct_change_message(form, formsets, add):
  471. """
  472. Construct a JSON structure describing changes from a changed object.
  473. Translations are deactivated so that strings are stored untranslated.
  474. Translation happens later on LogEntry access.
  475. """
  476. # Evaluating `form.changed_data` prior to disabling translations is required
  477. # to avoid fields affected by localization from being included incorrectly,
  478. # e.g. where date formats differ such as MM/DD/YYYY vs DD/MM/YYYY.
  479. changed_data = form.changed_data
  480. with translation_override(None):
  481. # Deactivate translations while fetching verbose_name for form
  482. # field labels and using `field_name`, if verbose_name is not provided.
  483. # Translations will happen later on LogEntry access.
  484. changed_field_labels = _get_changed_field_labels_from_form(form, changed_data)
  485. change_message = []
  486. if add:
  487. change_message.append({"added": {}})
  488. elif form.changed_data:
  489. change_message.append({"changed": {"fields": changed_field_labels}})
  490. if formsets:
  491. with translation_override(None):
  492. for formset in formsets:
  493. for added_object in formset.new_objects:
  494. change_message.append(
  495. {
  496. "added": {
  497. "name": str(added_object._meta.verbose_name),
  498. "object": str(added_object),
  499. }
  500. }
  501. )
  502. for changed_object, changed_fields in formset.changed_objects:
  503. change_message.append(
  504. {
  505. "changed": {
  506. "name": str(changed_object._meta.verbose_name),
  507. "object": str(changed_object),
  508. "fields": _get_changed_field_labels_from_form(
  509. formset.forms[0], changed_fields
  510. ),
  511. }
  512. }
  513. )
  514. for deleted_object in formset.deleted_objects:
  515. change_message.append(
  516. {
  517. "deleted": {
  518. "name": str(deleted_object._meta.verbose_name),
  519. "object": str(deleted_object),
  520. }
  521. }
  522. )
  523. return change_message
  524. def _get_changed_field_labels_from_form(form, changed_data):
  525. changed_field_labels = []
  526. for field_name in changed_data:
  527. try:
  528. verbose_field_name = form.fields[field_name].label or field_name
  529. except KeyError:
  530. verbose_field_name = field_name
  531. changed_field_labels.append(str(verbose_field_name))
  532. return changed_field_labels