LITE-12681 Added flag CQRS_SELECT_FOR_UPDATE to ReplicaMixin to suppo…

…rt automatic locking of records for CQRS save operations

dj_cqrs/managers.py

@@ -45,7 +45,11 @@ def save_instance(self, master_data, previous_data=None, sync=False):
             pk_value = mapped_data[pk_name]
             f_kwargs = {pk_name: pk_value}
 
-            instance = self.model._default_manager.filter(**f_kwargs).first()
+            qs = self.model._default_manager.filter(**f_kwargs)
+            if self.model.CQRS_SELECT_FOR_UPDATE:
+                qs = qs.select_for_update()
+
+            instance = qs.first()
 
             if instance:
                 return self.update_instance(

dj_cqrs/mixins.py

@@ -273,6 +273,9 @@ class ReplicaMixin(Model, metaclass=ReplicaMeta):
     CQRS_CUSTOM_SERIALIZATION = False
     """Set it to True to skip default data check."""
 
+    CQRS_SELECT_FOR_UPDATE = False
+    """Set it to True to acquire lock on instance creation/update."""
+
     objects = Manager()
 
     cqrs = ReplicaManager()

tests/dj_replica/models.py

@@ -48,6 +48,13 @@ class BadMappingModelRef(ReplicaMixin, models.Model):
     name = models.CharField(max_length=200)
 
 
+class LockModelRef(ReplicaMixin, models.Model):
+    CQRS_ID = 'lock'
+    CQRS_SELECT_FOR_UPDATE = True
+
+    id = models.IntegerField(primary_key=True)
+
+
 class Event(models.Model):
     pid = models.IntegerField()
     cqrs_id = models.CharField(max_length=20)

tests/test_replica/test_mixin.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from django.db.models import CharField, IntegerField
+from django.db.models import CharField, IntegerField, QuerySet
 from django.utils.timezone import now
 
 from dj_cqrs.metas import ReplicaMeta
@@ -469,3 +469,19 @@ def test_tracked_fields_mapped(mocker):
     _, kwargs = cqrs_update_mock.call_args
     assert 'previous_data' in kwargs
     assert kwargs['previous_data'] == {'name': 'text'}
+
+
+@pytest.mark.django_db
+def test_select_for_update_lock(mocker):
+    m = mocker.patch.object(
+        QuerySet, 'select_for_update', return_value=models.LockModelRef.objects.all(),
+    )
+
+    instance = models.LockModelRef.cqrs_save({
+        'id': 1,
+        'cqrs_revision': 0,
+        'cqrs_updated': now(),
+    })
+
+    assert instance.id == 1
+    m.assert_called_once()