import copy
import functools
import hashlib
import logging
import time
from collections import OrderedDict
from functools import update_wrapper
from django.db import models
from django.db.models import signals
from django.db.models.fields import Field
from django.db.models.signals import class_prepared, post_migrate
from django.utils.encoding import force_str
from django.utils.translation import gettext_lazy as _
from concurrency import forms
from concurrency.api import get_revision_of_object
from concurrency.config import conf
from concurrency.core import ConcurrencyOptions
from concurrency.utils import fqn, refetch
from .triggers import _TRIGGERS
logger = logging.getLogger(__name__)
OFFSET = int(time.mktime((2000, 1, 1, 0, 0, 0, 0, 0, 0)))
def class_prepared_concurrency_handler(sender, **kwargs):
if hasattr(sender, '_concurrencymeta'):
if sender != sender._concurrencymeta.base:
origin = getattr(sender._concurrencymeta.base, '_concurrencymeta')
local = copy.deepcopy(origin)
setattr(sender, '_concurrencymeta', local)
if hasattr(sender, 'ConcurrencyMeta'):
sender._concurrencymeta.enabled = getattr(sender.ConcurrencyMeta, 'enabled', True)
check_fields = getattr(sender.ConcurrencyMeta, 'check_fields', None)
ignore_fields = getattr(sender.ConcurrencyMeta, 'ignore_fields', None)
if check_fields and ignore_fields:
raise ValueError("Cannot set both 'check_fields' and 'ignore_fields'")
sender._concurrencymeta.check_fields = check_fields
sender._concurrencymeta.ignore_fields = ignore_fields
sender._concurrencymeta.increment = getattr(sender.ConcurrencyMeta, 'increment', True)
sender._concurrencymeta.skip = False
if not (sender._concurrencymeta.manually):
sender._concurrencymeta.field.wrap_model(sender)
setattr(sender, 'get_concurrency_version', get_revision_of_object)
def post_syncdb_concurrency_handler(sender, **kwargs):
from django.db import connections
from concurrency.triggers import create_triggers
databases = [alias for alias in connections]
create_triggers(databases)
class_prepared.connect(class_prepared_concurrency_handler,
dispatch_uid='class_prepared_concurrency_handler')
if conf.AUTO_CREATE_TRIGGERS:
post_migrate.connect(post_syncdb_concurrency_handler,
dispatch_uid='post_syncdb_concurrency_handler')
[docs]class VersionField(Field):
""" Base class """
def __init__(self, *args, **kwargs):
verbose_name = kwargs.get('verbose_name', None)
name = kwargs.get('name', None)
db_tablespace = kwargs.get('db_tablespace', None)
db_column = kwargs.get('db_column', None)
help_text = kwargs.get('help_text', _('record revision number'))
super().__init__(verbose_name, name,
help_text=help_text,
default=0,
db_tablespace=db_tablespace,
db_column=db_column)
def get_internal_type(self):
return "BigIntegerField"
def to_python(self, value):
return int(value)
def validate(self, value, model_instance):
pass
def formfield(self, **kwargs):
kwargs['form_class'] = self.form_class
kwargs['widget'] = forms.VersionField.widget
return super().formfield(**kwargs)
def contribute_to_class(self, cls, *args, **kwargs):
super().contribute_to_class(cls, *args, **kwargs)
if hasattr(cls, '_concurrencymeta') or cls._meta.abstract:
return
setattr(cls, '_concurrencymeta', ConcurrencyOptions())
cls._concurrencymeta.field = self
cls._concurrencymeta.base = cls
cls._concurrencymeta.triggers = []
def _set_version_value(self, model_instance, value):
setattr(model_instance, self.attname, int(value))
def pre_save(self, model_instance, add):
if add:
value = self._get_next_version(model_instance)
self._set_version_value(model_instance, value)
return getattr(model_instance, self.attname)
@classmethod
def wrap_model(cls, model, force=False):
if not force and model._concurrencymeta.versioned_save:
return
cls._wrap_model_methods(model)
model._concurrencymeta.versioned_save = True
@staticmethod
def _wrap_model_methods(model):
old_do_update = getattr(model, '_do_update')
setattr(model, '_do_update', model._concurrencymeta.field._wrap_do_update(old_do_update))
def _wrap_do_update(self, func):
def _do_update(model_instance, base_qs, using, pk_val, values, update_fields, forced_update):
version_field = model_instance._concurrencymeta.field
old_version = get_revision_of_object(model_instance)
if not version_field.model._meta.abstract:
if version_field.model is not base_qs.model:
return func(model_instance, base_qs, using, pk_val, values, update_fields, forced_update)
for i, (field, _1, value) in enumerate(values):
if field == version_field:
if (model_instance._concurrencymeta.increment and not
getattr(model_instance, '_concurrency_disable_increment', False)):
new_version = field._get_next_version(model_instance)
values[i] = (field, _1, new_version)
field._set_version_value(model_instance, new_version)
# else:
# new_version = old_version
break
# This provides a default if either (1) no values were provided or (2) we reached this code as part of a
# create. We don't need to worry about a race condition because a competing create should produce an
# error anyway.
updated = base_qs.filter(pk=pk_val).exists()
# This second situation can occur because `Model.save_base` calls `Model._save_parent` without relaying
# the `force_insert` flag that marks the process as a create. Eventually, `Model._save_table` will call
# this function as-if it were in the middle of an update. The update is expected to fail because there
# is no object to update and the caller will fall back on the create logic instead. We need to ensure
# the update fails (but does not raise an exception) under this circumstance by skipping the concurrency
# logic.
if values and updated:
if (model_instance._concurrencymeta.enabled and
conf.ENABLED and
not getattr(model_instance, '_concurrency_disabled', False) and
(old_version or conf.VERSION_FIELD_REQUIRED)):
filter_kwargs = {'pk': pk_val, version_field.attname: old_version}
updated = base_qs.filter(**filter_kwargs)._update(values) >= 1
if not updated:
version_field._set_version_value(model_instance, old_version)
updated = conf._callback(model_instance)
else:
filter_kwargs = {'pk': pk_val}
updated = base_qs.filter(**filter_kwargs)._update(values) >= 1
return updated
return update_wrapper(_do_update, func)
[docs]class IntegerVersionField(VersionField):
"""
Version Field that returns a "unique" version number for the record.
The version number is produced using time.time() * 1000000, to get the benefits
of microsecond if the system clock provides them.
"""
form_class = forms.VersionField
def _get_next_version(self, model_instance):
old_value = getattr(model_instance, self.attname, 0)
return max(int(old_value) + 1, (int(time.time() * 1000000) - OFFSET))
[docs]class AutoIncVersionField(VersionField):
"""
Version Field increment the revision number each commit
"""
form_class = forms.VersionField
def _get_next_version(self, model_instance):
return int(getattr(model_instance, self.attname, 0)) + 1
class TriggerVersionField(VersionField):
"""
Version Field increment the revision number each commit
"""
form_class = forms.VersionField
def __init__(self, *args, **kwargs):
self._trigger_name = kwargs.pop('trigger_name', None)
self._trigger_exists = False
super().__init__(*args, **kwargs)
def contribute_to_class(self, cls, *args, **kwargs):
super().contribute_to_class(cls, *args, **kwargs)
if not cls._meta.abstract or cls._meta.proxy:
if self not in _TRIGGERS:
_TRIGGERS.append(self)
def check(self, **kwargs):
errors = []
model = self.model
from django.core.checks import Warning
from django.db import connections, router
from concurrency.triggers import factory
alias = router.db_for_write(model)
connection = connections[alias]
f = factory(connection)
if not f.get_trigger(self):
errors.append(
Warning(
'Missed trigger for field {}'.format(self),
hint=None,
obj=None,
id='concurrency.W001',
)
)
return errors
@property
def trigger_name(self):
from concurrency.triggers import get_trigger_name
return get_trigger_name(self)
def _get_next_version(self, model_instance):
# always returns the same value
return int(getattr(model_instance, self.attname, 1))
def pre_save(self, model_instance, add):
# always returns the same value
return 1
@staticmethod
def _increment_version_number(obj):
old_value = get_revision_of_object(obj)
setattr(obj, obj._concurrencymeta.field.attname, int(old_value) + 1)
@staticmethod
def _wrap_model_methods(model):
super(TriggerVersionField, TriggerVersionField)._wrap_model_methods(model)
old_save = getattr(model, 'save')
setattr(model, 'save', model._concurrencymeta.field._wrap_save(old_save))
@staticmethod
def _wrap_save(func):
def inner(self, force_insert=False, force_update=False, using=None, **kwargs):
reload = kwargs.pop('refetch', False)
ret = func(self, force_insert, force_update, using, **kwargs)
TriggerVersionField._increment_version_number(self)
if reload:
ret = refetch(self)
setattr(self,
self._concurrencymeta.field.attname,
get_revision_of_object(ret))
return ret
return update_wrapper(inner, func)
def filter_fields(instance, field):
if not field.concrete:
# reverse relation
return False
if field.is_relation and field.related_model is None:
# generic foreignkeys
return False
if field.many_to_many and instance.pk is None:
# can't load remote object yet
return False
return True
class ConditionalVersionField(AutoIncVersionField):
def contribute_to_class(self, cls, *args, **kwargs):
super().contribute_to_class(cls, *args, **kwargs)
signals.post_init.connect(self._load_model,
sender=cls,
dispatch_uid=fqn(cls))
signals.post_save.connect(self._save_model,
sender=cls,
dispatch_uid=fqn(cls))
def _load_model(self, *args, **kwargs):
instance = kwargs['instance']
instance._concurrencymeta.initial = self._get_hash(instance)
def _save_model(self, *args, **kwargs):
instance = kwargs['instance']
instance._concurrencymeta.initial = self._get_hash(instance)
def _get_hash(self, instance):
values = OrderedDict()
opts = instance._meta
check_fields = instance._concurrencymeta.check_fields
ignore_fields = instance._concurrencymeta.ignore_fields
filter_ = functools.partial(filter_fields, instance)
if check_fields is None and ignore_fields is None:
fields = sorted([f.name for f in filter(filter_, instance._meta.get_fields())])
elif check_fields is None:
fields = sorted([f.name for f in filter(filter_, instance._meta.get_fields())
if f.name not in ignore_fields])
else:
fields = instance._concurrencymeta.check_fields
for field_name in fields:
# do not use getattr here. we do not need extra sql to retrieve
# FK. the raw value of the FK is enough
field = opts.get_field(field_name)
if isinstance(field, models.ManyToManyField):
values[field_name] = getattr(instance, field_name).values_list('pk', flat=True)
else:
values[field_name] = field.value_from_object(instance)
return hashlib.sha1(force_str(values).encode('utf-8')).hexdigest()
def _get_next_version(self, model_instance):
if not model_instance.pk:
return int(getattr(model_instance, self.attname) + 1)
old = model_instance._concurrencymeta.initial
new = self._get_hash(model_instance)
if old != new:
return int(getattr(model_instance, self.attname, 0) + 1)
return int(getattr(model_instance, self.attname, 0))