main.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631
  1. import warnings
  2. from datetime import datetime, timedelta
  3. from django import forms
  4. from django.conf import settings
  5. from django.contrib import messages
  6. from django.contrib.admin import FieldListFilter
  7. from django.contrib.admin.exceptions import (
  8. DisallowedModelAdminLookup,
  9. DisallowedModelAdminToField,
  10. )
  11. from django.contrib.admin.options import (
  12. IS_FACETS_VAR,
  13. IS_POPUP_VAR,
  14. TO_FIELD_VAR,
  15. IncorrectLookupParameters,
  16. ShowFacets,
  17. )
  18. from django.contrib.admin.utils import (
  19. build_q_object_from_lookup_parameters,
  20. get_fields_from_path,
  21. lookup_spawns_duplicates,
  22. prepare_lookup_value,
  23. quote,
  24. )
  25. from django.core.exceptions import (
  26. FieldDoesNotExist,
  27. ImproperlyConfigured,
  28. SuspiciousOperation,
  29. )
  30. from django.core.paginator import InvalidPage
  31. from django.db.models import F, Field, ManyToOneRel, OrderBy
  32. from django.db.models.constants import LOOKUP_SEP
  33. from django.db.models.expressions import Combinable
  34. from django.urls import reverse
  35. from django.utils.deprecation import RemovedInDjango60Warning
  36. from django.utils.http import urlencode
  37. from django.utils.inspect import func_supports_parameter
  38. from django.utils.timezone import make_aware
  39. from django.utils.translation import gettext
  40. # Changelist settings
  41. ALL_VAR = "all"
  42. ORDER_VAR = "o"
  43. PAGE_VAR = "p"
  44. SEARCH_VAR = "q"
  45. ERROR_FLAG = "e"
  46. IGNORED_PARAMS = (
  47. ALL_VAR,
  48. ORDER_VAR,
  49. SEARCH_VAR,
  50. IS_FACETS_VAR,
  51. IS_POPUP_VAR,
  52. TO_FIELD_VAR,
  53. )
  54. class ChangeListSearchForm(forms.Form):
  55. def __init__(self, *args, **kwargs):
  56. super().__init__(*args, **kwargs)
  57. # Populate "fields" dynamically because SEARCH_VAR is a variable:
  58. self.fields = {
  59. SEARCH_VAR: forms.CharField(required=False, strip=False),
  60. }
  61. class ChangeList:
  62. search_form_class = ChangeListSearchForm
  63. def __init__(
  64. self,
  65. request,
  66. model,
  67. list_display,
  68. list_display_links,
  69. list_filter,
  70. date_hierarchy,
  71. search_fields,
  72. list_select_related,
  73. list_per_page,
  74. list_max_show_all,
  75. list_editable,
  76. model_admin,
  77. sortable_by,
  78. search_help_text,
  79. ):
  80. self.model = model
  81. self.opts = model._meta
  82. self.lookup_opts = self.opts
  83. self.root_queryset = model_admin.get_queryset(request)
  84. self.list_display = list_display
  85. self.list_display_links = list_display_links
  86. self.list_filter = list_filter
  87. self.has_filters = None
  88. self.has_active_filters = None
  89. self.clear_all_filters_qs = None
  90. self.date_hierarchy = date_hierarchy
  91. self.search_fields = search_fields
  92. self.list_select_related = list_select_related
  93. self.list_per_page = list_per_page
  94. self.list_max_show_all = list_max_show_all
  95. self.model_admin = model_admin
  96. self.preserved_filters = model_admin.get_preserved_filters(request)
  97. self.sortable_by = sortable_by
  98. self.search_help_text = search_help_text
  99. # Get search parameters from the query string.
  100. _search_form = self.search_form_class(request.GET)
  101. if not _search_form.is_valid():
  102. for error in _search_form.errors.values():
  103. messages.error(request, ", ".join(error))
  104. self.query = _search_form.cleaned_data.get(SEARCH_VAR) or ""
  105. try:
  106. self.page_num = int(request.GET.get(PAGE_VAR, 1))
  107. except ValueError:
  108. self.page_num = 1
  109. self.show_all = ALL_VAR in request.GET
  110. self.is_popup = IS_POPUP_VAR in request.GET
  111. self.add_facets = model_admin.show_facets is ShowFacets.ALWAYS or (
  112. model_admin.show_facets is ShowFacets.ALLOW and IS_FACETS_VAR in request.GET
  113. )
  114. self.is_facets_optional = model_admin.show_facets is ShowFacets.ALLOW
  115. to_field = request.GET.get(TO_FIELD_VAR)
  116. if to_field and not model_admin.to_field_allowed(request, to_field):
  117. raise DisallowedModelAdminToField(
  118. "The field %s cannot be referenced." % to_field
  119. )
  120. self.to_field = to_field
  121. self.params = dict(request.GET.items())
  122. self.filter_params = dict(request.GET.lists())
  123. if PAGE_VAR in self.params:
  124. del self.params[PAGE_VAR]
  125. del self.filter_params[PAGE_VAR]
  126. if ERROR_FLAG in self.params:
  127. del self.params[ERROR_FLAG]
  128. del self.filter_params[ERROR_FLAG]
  129. self.remove_facet_link = self.get_query_string(remove=[IS_FACETS_VAR])
  130. self.add_facet_link = self.get_query_string({IS_FACETS_VAR: True})
  131. if self.is_popup:
  132. self.list_editable = ()
  133. else:
  134. self.list_editable = list_editable
  135. self.queryset = self.get_queryset(request)
  136. self.get_results(request)
  137. if self.is_popup:
  138. title = gettext("Select %s")
  139. elif self.model_admin.has_change_permission(request):
  140. title = gettext("Select %s to change")
  141. else:
  142. title = gettext("Select %s to view")
  143. self.title = title % self.opts.verbose_name
  144. self.pk_attname = self.lookup_opts.pk.attname
  145. def __repr__(self):
  146. return "<%s: model=%s model_admin=%s>" % (
  147. self.__class__.__qualname__,
  148. self.model.__qualname__,
  149. self.model_admin.__class__.__qualname__,
  150. )
  151. def get_filters_params(self, params=None):
  152. """
  153. Return all params except IGNORED_PARAMS.
  154. """
  155. params = params or self.filter_params
  156. lookup_params = params.copy() # a dictionary of the query string
  157. # Remove all the parameters that are globally and systematically
  158. # ignored.
  159. for ignored in IGNORED_PARAMS:
  160. if ignored in lookup_params:
  161. del lookup_params[ignored]
  162. return lookup_params
  163. def get_filters(self, request):
  164. lookup_params = self.get_filters_params()
  165. may_have_duplicates = False
  166. has_active_filters = False
  167. supports_request = func_supports_parameter(
  168. self.model_admin.lookup_allowed, "request"
  169. )
  170. if not supports_request:
  171. warnings.warn(
  172. f"`request` must be added to the signature of "
  173. f"{self.model_admin.__class__.__qualname__}.lookup_allowed().",
  174. RemovedInDjango60Warning,
  175. )
  176. for key, value_list in lookup_params.items():
  177. for value in value_list:
  178. params = (key, value, request) if supports_request else (key, value)
  179. if not self.model_admin.lookup_allowed(*params):
  180. raise DisallowedModelAdminLookup(f"Filtering by {key} not allowed")
  181. filter_specs = []
  182. for list_filter in self.list_filter:
  183. lookup_params_count = len(lookup_params)
  184. if callable(list_filter):
  185. # This is simply a custom list filter class.
  186. spec = list_filter(request, lookup_params, self.model, self.model_admin)
  187. else:
  188. field_path = None
  189. if isinstance(list_filter, (tuple, list)):
  190. # This is a custom FieldListFilter class for a given field.
  191. field, field_list_filter_class = list_filter
  192. else:
  193. # This is simply a field name, so use the default
  194. # FieldListFilter class that has been registered for the
  195. # type of the given field.
  196. field, field_list_filter_class = list_filter, FieldListFilter.create
  197. if not isinstance(field, Field):
  198. field_path = field
  199. field = get_fields_from_path(self.model, field_path)[-1]
  200. spec = field_list_filter_class(
  201. field,
  202. request,
  203. lookup_params,
  204. self.model,
  205. self.model_admin,
  206. field_path=field_path,
  207. )
  208. # field_list_filter_class removes any lookup_params it
  209. # processes. If that happened, check if duplicates should be
  210. # removed.
  211. if lookup_params_count > len(lookup_params):
  212. may_have_duplicates |= lookup_spawns_duplicates(
  213. self.lookup_opts,
  214. field_path,
  215. )
  216. if spec and spec.has_output():
  217. filter_specs.append(spec)
  218. if lookup_params_count > len(lookup_params):
  219. has_active_filters = True
  220. if self.date_hierarchy:
  221. # Create bounded lookup parameters so that the query is more
  222. # efficient.
  223. year = lookup_params.pop("%s__year" % self.date_hierarchy, None)
  224. if year is not None:
  225. month = lookup_params.pop("%s__month" % self.date_hierarchy, None)
  226. day = lookup_params.pop("%s__day" % self.date_hierarchy, None)
  227. try:
  228. from_date = datetime(
  229. int(year[-1]),
  230. int(month[-1] if month is not None else 1),
  231. int(day[-1] if day is not None else 1),
  232. )
  233. except ValueError as e:
  234. raise IncorrectLookupParameters(e) from e
  235. if day:
  236. to_date = from_date + timedelta(days=1)
  237. elif month:
  238. # In this branch, from_date will always be the first of a
  239. # month, so advancing 32 days gives the next month.
  240. to_date = (from_date + timedelta(days=32)).replace(day=1)
  241. else:
  242. to_date = from_date.replace(year=from_date.year + 1)
  243. if settings.USE_TZ:
  244. from_date = make_aware(from_date)
  245. to_date = make_aware(to_date)
  246. lookup_params.update(
  247. {
  248. "%s__gte" % self.date_hierarchy: [from_date],
  249. "%s__lt" % self.date_hierarchy: [to_date],
  250. }
  251. )
  252. # At this point, all the parameters used by the various ListFilters
  253. # have been removed from lookup_params, which now only contains other
  254. # parameters passed via the query string. We now loop through the
  255. # remaining parameters both to ensure that all the parameters are valid
  256. # fields and to determine if at least one of them spawns duplicates. If
  257. # the lookup parameters aren't real fields, then bail out.
  258. try:
  259. for key, value in lookup_params.items():
  260. lookup_params[key] = prepare_lookup_value(key, value)
  261. may_have_duplicates |= lookup_spawns_duplicates(self.lookup_opts, key)
  262. return (
  263. filter_specs,
  264. bool(filter_specs),
  265. lookup_params,
  266. may_have_duplicates,
  267. has_active_filters,
  268. )
  269. except FieldDoesNotExist as e:
  270. raise IncorrectLookupParameters(e) from e
  271. def get_query_string(self, new_params=None, remove=None):
  272. if new_params is None:
  273. new_params = {}
  274. if remove is None:
  275. remove = []
  276. p = self.filter_params.copy()
  277. for r in remove:
  278. for k in list(p):
  279. if k.startswith(r):
  280. del p[k]
  281. for k, v in new_params.items():
  282. if v is None:
  283. if k in p:
  284. del p[k]
  285. else:
  286. p[k] = v
  287. return "?%s" % urlencode(sorted(p.items()), doseq=True)
  288. def get_results(self, request):
  289. paginator = self.model_admin.get_paginator(
  290. request, self.queryset, self.list_per_page
  291. )
  292. # Get the number of objects, with admin filters applied.
  293. result_count = paginator.count
  294. # Get the total number of objects, with no admin filters applied.
  295. # Note this isn't necessarily the same as result_count in the case of
  296. # no filtering. Filters defined in list_filters may still apply some
  297. # default filtering which may be removed with query parameters.
  298. if self.model_admin.show_full_result_count:
  299. full_result_count = self.root_queryset.count()
  300. else:
  301. full_result_count = None
  302. can_show_all = result_count <= self.list_max_show_all
  303. multi_page = result_count > self.list_per_page
  304. # Get the list of objects to display on this page.
  305. if (self.show_all and can_show_all) or not multi_page:
  306. result_list = self.queryset._clone()
  307. else:
  308. try:
  309. result_list = paginator.page(self.page_num).object_list
  310. except InvalidPage:
  311. raise IncorrectLookupParameters
  312. self.result_count = result_count
  313. self.show_full_result_count = self.model_admin.show_full_result_count
  314. # Admin actions are shown if there is at least one entry
  315. # or if entries are not counted because show_full_result_count is disabled
  316. self.show_admin_actions = not self.show_full_result_count or bool(
  317. full_result_count
  318. )
  319. self.full_result_count = full_result_count
  320. self.result_list = result_list
  321. self.can_show_all = can_show_all
  322. self.multi_page = multi_page
  323. self.paginator = paginator
  324. def _get_default_ordering(self):
  325. ordering = []
  326. if self.model_admin.ordering:
  327. ordering = self.model_admin.ordering
  328. elif self.lookup_opts.ordering:
  329. ordering = self.lookup_opts.ordering
  330. return ordering
  331. def get_ordering_field(self, field_name):
  332. """
  333. Return the proper model field name corresponding to the given
  334. field_name to use for ordering. field_name may either be the name of a
  335. proper model field, possibly across relations, or the name of a method
  336. (on the admin or model) or a callable with the 'admin_order_field'
  337. attribute. Return None if no proper model field name can be matched.
  338. """
  339. try:
  340. field = self.lookup_opts.get_field(field_name)
  341. return field.name
  342. except FieldDoesNotExist:
  343. # See whether field_name is a name of a non-field
  344. # that allows sorting.
  345. if callable(field_name):
  346. attr = field_name
  347. elif hasattr(self.model_admin, field_name):
  348. attr = getattr(self.model_admin, field_name)
  349. else:
  350. try:
  351. attr = getattr(self.model, field_name)
  352. except AttributeError:
  353. if LOOKUP_SEP in field_name:
  354. return field_name
  355. raise
  356. if isinstance(attr, property) and hasattr(attr, "fget"):
  357. attr = attr.fget
  358. return getattr(attr, "admin_order_field", None)
  359. def get_ordering(self, request, queryset):
  360. """
  361. Return the list of ordering fields for the change list.
  362. First check the get_ordering() method in model admin, then check
  363. the object's default ordering. Then, any manually-specified ordering
  364. from the query string overrides anything. Finally, a deterministic
  365. order is guaranteed by calling _get_deterministic_ordering() with the
  366. constructed ordering.
  367. """
  368. params = self.params
  369. ordering = list(
  370. self.model_admin.get_ordering(request) or self._get_default_ordering()
  371. )
  372. if ORDER_VAR in params:
  373. # Clear ordering and used params
  374. ordering = []
  375. order_params = params[ORDER_VAR].split(".")
  376. for p in order_params:
  377. try:
  378. none, pfx, idx = p.rpartition("-")
  379. field_name = self.list_display[int(idx)]
  380. order_field = self.get_ordering_field(field_name)
  381. if not order_field:
  382. continue # No 'admin_order_field', skip it
  383. if isinstance(order_field, OrderBy):
  384. if pfx == "-":
  385. order_field = order_field.copy()
  386. order_field.reverse_ordering()
  387. ordering.append(order_field)
  388. elif hasattr(order_field, "resolve_expression"):
  389. # order_field is an expression.
  390. ordering.append(
  391. order_field.desc() if pfx == "-" else order_field.asc()
  392. )
  393. # reverse order if order_field has already "-" as prefix
  394. elif pfx == "-" and order_field.startswith(pfx):
  395. ordering.append(order_field.removeprefix(pfx))
  396. else:
  397. ordering.append(pfx + order_field)
  398. except (IndexError, ValueError):
  399. continue # Invalid ordering specified, skip it.
  400. # Add the given query's ordering fields, if any.
  401. ordering.extend(queryset.query.order_by)
  402. return self._get_deterministic_ordering(ordering)
  403. def _get_deterministic_ordering(self, ordering):
  404. """
  405. Ensure a deterministic order across all database backends. Search for a
  406. single field or unique together set of fields providing a total
  407. ordering. If these are missing, augment the ordering with a descendant
  408. primary key.
  409. """
  410. ordering = list(ordering)
  411. ordering_fields = set()
  412. total_ordering_fields = {"pk"} | {
  413. field.attname
  414. for field in self.lookup_opts.fields
  415. if field.unique and not field.null
  416. }
  417. for part in ordering:
  418. # Search for single field providing a total ordering.
  419. field_name = None
  420. if isinstance(part, str):
  421. field_name = part.lstrip("-")
  422. elif isinstance(part, F):
  423. field_name = part.name
  424. elif isinstance(part, OrderBy) and isinstance(part.expression, F):
  425. field_name = part.expression.name
  426. if field_name:
  427. # Normalize attname references by using get_field().
  428. try:
  429. field = self.lookup_opts.get_field(field_name)
  430. except FieldDoesNotExist:
  431. # Could be "?" for random ordering or a related field
  432. # lookup. Skip this part of introspection for now.
  433. continue
  434. # Ordering by a related field name orders by the referenced
  435. # model's ordering. Skip this part of introspection for now.
  436. if field.remote_field and field_name == field.name:
  437. continue
  438. if field.attname in total_ordering_fields:
  439. break
  440. ordering_fields.add(field.attname)
  441. else:
  442. # No single total ordering field, try unique_together and total
  443. # unique constraints.
  444. constraint_field_names = (
  445. *self.lookup_opts.unique_together,
  446. *(
  447. constraint.fields
  448. for constraint in self.lookup_opts.total_unique_constraints
  449. ),
  450. )
  451. for field_names in constraint_field_names:
  452. # Normalize attname references by using get_field().
  453. fields = [
  454. self.lookup_opts.get_field(field_name) for field_name in field_names
  455. ]
  456. # Composite unique constraints containing a nullable column
  457. # cannot ensure total ordering.
  458. if any(field.null for field in fields):
  459. continue
  460. if ordering_fields.issuperset(field.attname for field in fields):
  461. break
  462. else:
  463. # If no set of unique fields is present in the ordering, rely
  464. # on the primary key to provide total ordering.
  465. ordering.append("-pk")
  466. return ordering
  467. def get_ordering_field_columns(self):
  468. """
  469. Return a dictionary of ordering field column numbers and asc/desc.
  470. """
  471. # We must cope with more than one column having the same underlying sort
  472. # field, so we base things on column numbers.
  473. ordering = self._get_default_ordering()
  474. ordering_fields = {}
  475. if ORDER_VAR not in self.params:
  476. # for ordering specified on ModelAdmin or model Meta, we don't know
  477. # the right column numbers absolutely, because there might be more
  478. # than one column associated with that ordering, so we guess.
  479. for field in ordering:
  480. if isinstance(field, (Combinable, OrderBy)):
  481. if not isinstance(field, OrderBy):
  482. field = field.asc()
  483. if isinstance(field.expression, F):
  484. order_type = "desc" if field.descending else "asc"
  485. field = field.expression.name
  486. else:
  487. continue
  488. elif field.startswith("-"):
  489. field = field.removeprefix("-")
  490. order_type = "desc"
  491. else:
  492. order_type = "asc"
  493. for index, attr in enumerate(self.list_display):
  494. if self.get_ordering_field(attr) == field:
  495. ordering_fields[index] = order_type
  496. break
  497. else:
  498. for p in self.params[ORDER_VAR].split("."):
  499. none, pfx, idx = p.rpartition("-")
  500. try:
  501. idx = int(idx)
  502. except ValueError:
  503. continue # skip it
  504. ordering_fields[idx] = "desc" if pfx == "-" else "asc"
  505. return ordering_fields
  506. def get_queryset(self, request, exclude_parameters=None):
  507. # First, we collect all the declared list filters.
  508. (
  509. self.filter_specs,
  510. self.has_filters,
  511. remaining_lookup_params,
  512. filters_may_have_duplicates,
  513. self.has_active_filters,
  514. ) = self.get_filters(request)
  515. # Then, we let every list filter modify the queryset to its liking.
  516. qs = self.root_queryset
  517. for filter_spec in self.filter_specs:
  518. if (
  519. exclude_parameters is None
  520. or filter_spec.expected_parameters() != exclude_parameters
  521. ):
  522. new_qs = filter_spec.queryset(request, qs)
  523. if new_qs is not None:
  524. qs = new_qs
  525. try:
  526. # Finally, we apply the remaining lookup parameters from the query
  527. # string (i.e. those that haven't already been processed by the
  528. # filters).
  529. q_object = build_q_object_from_lookup_parameters(remaining_lookup_params)
  530. qs = qs.filter(q_object)
  531. except (SuspiciousOperation, ImproperlyConfigured):
  532. # Allow certain types of errors to be re-raised as-is so that the
  533. # caller can treat them in a special way.
  534. raise
  535. except Exception as e:
  536. # Every other error is caught with a naked except, because we don't
  537. # have any other way of validating lookup parameters. They might be
  538. # invalid if the keyword arguments are incorrect, or if the values
  539. # are not in the correct type, so we might get FieldError,
  540. # ValueError, ValidationError, or ?.
  541. raise IncorrectLookupParameters(e)
  542. if not qs.query.select_related:
  543. qs = self.apply_select_related(qs)
  544. # Set ordering.
  545. ordering = self.get_ordering(request, qs)
  546. qs = qs.order_by(*ordering)
  547. # Apply search results
  548. qs, search_may_have_duplicates = self.model_admin.get_search_results(
  549. request,
  550. qs,
  551. self.query,
  552. )
  553. # Set query string for clearing all filters.
  554. self.clear_all_filters_qs = self.get_query_string(
  555. new_params=remaining_lookup_params,
  556. remove=self.get_filters_params(),
  557. )
  558. # Remove duplicates from results, if necessary
  559. if filters_may_have_duplicates | search_may_have_duplicates:
  560. return qs.distinct()
  561. else:
  562. return qs
  563. def apply_select_related(self, qs):
  564. if self.list_select_related is True:
  565. return qs.select_related()
  566. if self.list_select_related is False:
  567. if self.has_related_field_in_list_display():
  568. return qs.select_related()
  569. if self.list_select_related:
  570. return qs.select_related(*self.list_select_related)
  571. return qs
  572. def has_related_field_in_list_display(self):
  573. for field_name in self.list_display:
  574. try:
  575. field = self.lookup_opts.get_field(field_name)
  576. except FieldDoesNotExist:
  577. pass
  578. else:
  579. if isinstance(field.remote_field, ManyToOneRel):
  580. # <FK>_id field names don't require a join.
  581. if field_name != field.attname:
  582. return True
  583. return False
  584. def url_for_result(self, result):
  585. pk = getattr(result, self.pk_attname)
  586. return reverse(
  587. "admin:%s_%s_change" % (self.opts.app_label, self.opts.model_name),
  588. args=(quote(pk),),
  589. current_app=self.model_admin.admin_site.name,
  590. )