import operator
import re
from functools import reduce
import django
from django.contrib import admin, messages
from django.contrib.admin import helpers
from django.core.checks import Error
from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.db.models import Q
from django.forms.formsets import (INITIAL_FORM_COUNT, MAX_NUM_FORM_COUNT,
TOTAL_FORM_COUNT, ManagementForm,)
from django.forms.models import BaseModelFormSet
from django.http import HttpResponse, HttpResponseRedirect
from django.utils.encoding import force_str
from django.utils.safestring import mark_safe
from django.utils.translation import ngettext
from concurrency import core, forms
from concurrency.api import get_revision_of_object
from concurrency.compat import concurrency_param_name
from concurrency.config import CONCURRENCY_LIST_EDITABLE_POLICY_ABORT_ALL, conf
from concurrency.exceptions import RecordModifiedError
from concurrency.forms import ConcurrentForm, VersionWidget
from concurrency.utils import flatten
ALL = object()
[docs]class ConcurrencyActionMixin:
check_concurrent_action = True
def action_checkbox(self, obj):
"""
A list_display column containing a checkbox widget.
"""
if self.check_concurrent_action:
return helpers.checkbox.render(helpers.ACTION_CHECKBOX_NAME,
force_str("%s,%s" % (obj.pk, get_revision_of_object(obj))))
else: # pragma: no cover
return super().action_checkbox(obj)
action_checkbox.short_description = mark_safe('<input type="checkbox" id="action-toggle" />')
action_checkbox.allow_tags = True
def get_confirmation_template(self):
return "concurrency/delete_selected_confirmation.html"
def response_action(self, request, queryset): # noqa
"""
Handle an admin action. This is called if a request is POSTed to the
changelist; it returns an HttpResponse if the action was handled, and
None otherwise.
"""
# There can be multiple action forms on the page (at the top
# and bottom of the change list, for example). Get the action
# whose button was pushed.
try:
action_index = int(request.POST.get('index', 0))
except ValueError: # pragma: no cover
action_index = 0
# Construct the action form.
data = request.POST.copy()
data.pop(helpers.ACTION_CHECKBOX_NAME, None)
data.pop("index", None)
# Use the action whose button was pushed
try:
data.update({'action': data.getlist('action')[action_index]})
except IndexError: # pragma: no cover
# If we didn't get an action from the chosen form that's invalid
# POST data, so by deleting action it'll fail the validation check
# below. So no need to do anything here
pass
action_form = self.action_form(data, auto_id=None)
action_form.fields['action'].choices = self.get_action_choices(request)
# If the form's valid we can handle the action.
if action_form.is_valid():
action = action_form.cleaned_data['action']
func, name, description = self.get_actions(request)[action]
# Get the list of selected PKs. If nothing's selected, we can't
# perform an action on it, so bail.
if action_form.cleaned_data['select_across']:
selected = ALL
else:
selected = request.POST.getlist(helpers.ACTION_CHECKBOX_NAME)
if not selected:
return None
revision_field = self.model._concurrencymeta.field
if self.check_concurrent_action:
self.delete_selected_confirmation_template = self.get_confirmation_template()
# If select_across we have to avoid the use of concurrency
if selected is not ALL:
filters = []
for x in selected:
try:
pk, version = x.split(",")
except ValueError: # pragma: no cover
raise ImproperlyConfigured('`ConcurrencyActionMixin` error.'
'A tuple with `primary_key, version_number` '
'expected: `%s` found' % x)
filters.append(Q(**{'pk': pk,
revision_field.attname: version}))
queryset = queryset.filter(reduce(operator.or_, filters))
if len(selected) != queryset.count():
messages.error(request, 'One or more record were updated. '
'(Probably by other user) '
'The execution was aborted.')
return HttpResponseRedirect(".")
else:
messages.warning(request, 'Selecting all records, you will avoid the concurrency check')
response = func(self, request, queryset)
# Actions may return an HttpResponse, which will be used as the
# response from the POST. If not, we'll be a good little HTTP
# citizen and redirect back to the changelist page.
if isinstance(response, HttpResponse):
return response
else:
return HttpResponseRedirect(".")
class ConcurrentManagementForm(ManagementForm):
def __init__(self, *args, **kwargs):
self._versions = kwargs.pop('versions', [])
super().__init__(*args, **kwargs)
def _get_concurrency_fields(self):
v = []
for pk, version in self._versions:
v.append(f'<input type="hidden" name="{concurrency_param_name}_{pk}" value="{version}">')
return mark_safe("".join(v))
def render(self, template_name=None, context=None, renderer=None):
out = super().render(template_name, context, renderer)
return out + self._get_concurrency_fields()
def __str__(self):
if django.VERSION[:2] >= (4, 0):
return self.render()
else:
return super().__str__()
__html__ = __str__
def _html_output(self, normal_row, error_row, row_ender, help_text_html, errors_on_separate_row):
ret = super()._html_output(normal_row, error_row, row_ender, help_text_html, errors_on_separate_row)
return mark_safe("{0}{1}".format(ret, self._get_concurrency_fields()))
class ConcurrentBaseModelFormSet(BaseModelFormSet):
def _management_form(self):
"""Returns the ManagementForm instance for this FormSet."""
if self.is_bound:
form = ConcurrentManagementForm(self.data, auto_id=self.auto_id,
prefix=self.prefix)
if not form.is_valid():
raise ValidationError('ManagementForm data is missing or has been tampered with')
else:
form = ConcurrentManagementForm(auto_id=self.auto_id,
prefix=self.prefix,
initial={TOTAL_FORM_COUNT: self.total_form_count(),
INITIAL_FORM_COUNT: self.initial_form_count(),
MAX_NUM_FORM_COUNT: self.max_num},
versions=[(form.instance.pk, get_revision_of_object(form.instance)) for form
in self.initial_forms])
return form
management_form = property(_management_form)
[docs]class ConcurrencyListEditableMixin:
list_editable_policy = conf.POLICY
def get_changelist_formset(self, request, **kwargs):
kwargs['formset'] = ConcurrentBaseModelFormSet
return super().get_changelist_formset(request, **kwargs)
def _add_conflict(self, request, obj):
if hasattr(request, '_concurrency_list_editable_errors'):
request._concurrency_list_editable_errors.append(obj.pk)
else:
request._concurrency_list_editable_errors = [obj.pk]
def _get_conflicts(self, request):
if hasattr(request, '_concurrency_list_editable_errors'):
return request._concurrency_list_editable_errors
else:
return []
def save_model(self, request, obj, form, change):
try:
if change:
version = request.POST.get(f'{concurrency_param_name}_{obj.pk}', None)
if version:
core._set_version(obj, version)
super().save_model(request, obj, form, change)
except RecordModifiedError:
self._add_conflict(request, obj)
# If policy is set to 'silent' the user will be informed using message_user
# raise Exception if not silent.
# NOTE:
# list_editable_policy MUST have the LIST_EDITABLE_POLICY_ABORT_ALL
# set to work properly
if self.list_editable_policy == CONCURRENCY_LIST_EDITABLE_POLICY_ABORT_ALL:
raise
def log_change(self, request, object, message):
if object.pk in self._get_conflicts(request):
return
return super().log_change(request, object, message)
def log_deletion(self, request, object, object_repr):
if object.pk in self._get_conflicts(request):
return
return super().log_deletion(request, object, object_repr)
def message_user(self, request, message, *args, **kwargs):
# This is ugly but we do not want to touch the changelist_view() code.
opts = self.model._meta
conflicts = self._get_conflicts(request)
if conflicts:
names = force_str(opts.verbose_name), force_str(opts.verbose_name_plural)
pattern = r"(?P<num>\d+) ({0}|{1})".format(*names)
rex = re.compile(pattern)
m = rex.match(message)
concurrency_errros = len(conflicts)
if m:
updated_record = int(m.group('num')) - concurrency_errros
ids = ",".join(map(str, conflicts))
messages.error(request,
ngettext("Record with pk `{0}` has been modified and was not updated",
"Records `{0}` have been modified and were not updated",
concurrency_errros).format(ids))
if updated_record == 1:
name = force_str(opts.verbose_name)
else:
name = force_str(opts.verbose_name_plural)
message = None
if updated_record > 0:
message = ngettext("%(count)s %(name)s was changed successfully.",
"%(count)s %(name)s were changed successfully.",
updated_record) % {'count': updated_record,
'name': name}
return super().message_user(request, message, *args, **kwargs)
[docs]class ConcurrentModelAdmin(ConcurrencyActionMixin,
ConcurrencyListEditableMixin,
admin.ModelAdmin):
form = ConcurrentForm
formfield_overrides = {forms.VersionField: {'widget': VersionWidget}}
def check(self, **kwargs):
errors = []
if self.fields:
version_field = self.model._concurrencymeta.field
if version_field.name not in self.fields:
errors.append(
Error(
'Missed version field in {} fields definition'.format(self),
hint="Please add '{}' to the 'fields' attribute".format(version_field.name),
obj=None,
id='concurrency.A001',
)
)
if self.fieldsets:
version_field = self.model._concurrencymeta.field
fields = flatten([v['fields'] for k, v in self.fieldsets])
if version_field.name not in fields:
errors.append(
Error(
'Missed version field in {} fieldsets definition'.format(self),
hint="Please add '{}' to the 'fieldsets' attribute".format(version_field.name),
obj=None,
id='concurrency.A002',
)
)
return errors