Repository: pynamodb/PynamoDB

Issue:
Model.Update should not inject version condition
# The problem

The `VersionAttribute` design pattern consists of
- **incrementing the version with every change** — to trip others so they don't overwrite your change
- **conditioning on the version** — to trip yourself so you don't overwrite others' change

There are two methods that can modify an existing model:
- `Model.save`
- `Model.update`

We've observed it to be a common pattern to change just a few fields and then call `Model.save`. In this case, a version condition is a good precaution to avoid overwriting more attributes than you meant to. Once users start seeing high % of conditional failures ("conflicts"), a more nuanced approach is called for, using `Model.update`. Often "last write wins" is good enough as long as you don't write unrelated attributes. Other times, a more nuanced condition can be used (e.g. update field X only if it had value Y before).

Unfortunately when we introduced `VersionAttribute`, we’ve made `Model.update` [append a condition](https://github.com/pynamodb/PynamoDB/blob/b5600ea8c44dca523f94a0d1c8491d7ee0a8b3f5/pynamodb/models.py#L425-L427) on the version attribute. This is probably counter-productive to what I described above.

# The proposal

In PynamoDB 6.0, we can allow changing this behavior.

Options:
1. Stop appending the condition in `Model.update`. This risks silently breaking existing apps.
2. Make it optional and default to append. This drags the bad behavior as a default into green-field apps.
3. Make it optional and default to not appending. This risks silently breaking existing apps.
4. Ask user to decide, provide no default. 📌 

I'm inclined towards option (4). Adding a parameter to `Model.update` is disruptive, and also risks adding a hidden crash, since not every `Model.update` call is necessarily exercised by tests. Instead, we can add a parameter to `VersionAttribute`, since it's executed at class load / module load, and likely prevent an app from starting up until its maintainer makes a conscious decision.

We should name it something like `legacy_update_behavior: bool` as to discourage setting it to `True` in green-field apps:
```diff
-version = VersionAttribute()
+version = VersionAttribute(legacy_update_behavior=False)
```

In future versions we could switch to Option 1 by removing it.

Discussion:
ikonst: @garrettheel What you reviewed before was totally broken -- when `False` it didn't add condition **nor** incremented version 😅 

This time
- fixed the thing
- updated tests
- removed the prose from VersionAttribute, instead linking to an "optimistic locking" page that we already had, and adding a section to it about skipping the condition

Patch:
diff --git a/docs/optimistic_locking.rst b/docs/optimistic_locking.rst
index 00fac5f..3c23592 100644
--- a/docs/optimistic_locking.rst
+++ b/docs/optimistic_locking.rst
@@ -1,3 +1,5 @@
+.. _optimistic_locking:
+
 ==================
 Optimistic Locking
 ==================
@@ -18,7 +20,16 @@ See also:
 Version Attribute
 -----------------
 
-To enable optimistic locking for a table simply add a ``VersionAttribute`` to your model definition.
+To enable optimistic locking for a table, add a ``VersionAttribute`` to your model definition. The presence of this
+attribute will change the model's behaviors:
+
+* :meth:`~pynamodb.models.Model.save` and :meth:`~pynamodb.models.Model.update` would increment the version attribute
+  every time the model is persisted. This allows concurrent updates not to overwrite each other, at the expense
+  of the latter update failing.
+* :meth:`~pynamodb.models.Model.save`, :meth:`~pynamodb.models.Model.update`
+  and :meth:`~pynamodb.models.Model.delete` would fail if they are the "latter update" (by adding to the update's
+  :ref:`conditions <conditional_operations>`). This behavior is optional since sometimes a more granular approach
+  can be desired (see :ref:`optimistic_locking_version_condition`).
 
 .. code-block:: python
 
@@ -83,7 +94,7 @@ These operations will fail if the local object is out-of-date.
       except TransactWriteError as e:
           assert isinstance(e.cause, ClientError)
           assert e.cause_response_code == "TransactionCanceledException"
-          assert "ConditionalCheckFailed" in e.cause_response_message
+          assert any(r.code == "ConditionalCheckFailed" for r in e.cancellation_reasons)
       else:
           raise AssertionError("The version attribute conditional check should have failed.")
 
@@ -104,6 +115,46 @@ These operations will fail if the local object is out-of-date.
   with assert_condition_check_fails():
       office.delete()
 
+
+.. _optimistic_locking_version_condition:
+
+Conditioning on the version
+---------------------------
+
+To have :meth:`~pynamodb.models.Model.save`, :meth:`~pynamodb.models.Model.update` or  :meth:`~pynamodb.models.Model.delete`
+execute even if the item was changed by someone else, pass the ``add_version_condition=False`` parameter.
+In this mode, updates would perform unconditionally but would still increment the version:
+in other words, you could make other updates fail, but your update will succeed.
+
+Done indiscriminately, this would be unsafe, but can be useful in certain scenarios:
+
+#. For ``save``, this is almost always unsafe and undesirable.
+#. For ``update``, use it when updating attributes for which a "last write wins" approach is acceptable,
+   or if you're otherwise conditioning the update in a way that is more domain-specific.
+#. For ``delete``, use it to delete the item regardless of its contents.
+
+For example, if your ``save`` operation experiences frequent "ConditionalCheckFailedException" failures,
+rewrite your code to call ``update`` with individual attributes while passing :code:`add_version_condition=False`.
+By disabling the version condition, you could no longer rely on the checks you've done prior to the modification (due to
+what is known as the "time-of-check to time-of-use" problem). Therefore, consider adding domain-specific conditions
+to ensure the item in the table is in the expected state prior to the update.
+
+For example, let's consider a hotel room-booking service with the conventional constraint that only one person
+can book a room at a time. We can switch from a ``save`` to an ``update`` by specifying the individual attributes
+and rewriting the `if` statement as a condition:
+
+    .. code-block:: diff
+
+        - if room.booked_by:
+        -   raise Exception("Room is already booked")
+        - room.booked_by = user_id
+        - room.save()
+        + room.update(
+        +   actions=[Room.booked_by.set(user_id)],
+        +   condition=Room.booked_by.does_not_exist(),
+        +   add_version_condition=False,
+        + )
+
 Transactions
 ------------
 
diff --git a/docs/release_notes.rst b/docs/release_notes.rst
index adffd87..4a4b0a7 100644
--- a/docs/release_notes.rst
+++ b/docs/release_notes.rst
@@ -11,12 +11,19 @@ This is a major release and contains breaking changes. Please read the notes bel
   * The attributes' internal encoding has changed. To prevent this change going unnoticed, :code:`legacy_encoding` have been made required: see :doc:`upgrading_binary` for details.
     If your codebase uses :py:class:`~pynamodb.attributes.BinaryAttribute` or :py:class:`~pynamodb.attributes.BinarySetAttribute`,
     go over the attribute declarations and mark them accordingly.
-  * When using binary attributes, the return value of :py:func:`~pynamodb.models.Model.serialize` will no longer be JSON-serializable
-    since it will contain :code:`bytes` objects. Use `:py:func:`~pynamodb.models.Model.to_dynamodb_dict`
+  * When using binary attributes, the return value of :meth:`~pynamodb.models.Model.serialize` will no longer be JSON-serializable
+    since it will contain :code:`bytes` objects. Use :meth:`~pynamodb.models.Model.to_dynamodb_dict`
     for a safe JSON-serializable representation.
 
 * Python 3.6 is no longer supported.
-* Index count, query, and scan methods are now instance methods.
+* :meth:`Index.count <pynamodb.indexes.Index.count>`, :meth:`Index.query <pynamodb.indexes.Index.query>`,
+  and :meth:`Indexn.scan <pynamodb.indexes.Index.scan>` are now instance methods.
+
+Other changes in this release:
+
+* :meth:`~pynamodb.models.Model.save`, :meth:`~pynamodb.models.Model.update`, :meth:`~pynamodb.models.Model.delete_item`,
+  and :meth:`~pynamodb.models.Model.delete` now accept a ``add_version_condition`` parameter.
+  See :ref:`optimistic_locking_version_condition` for more details.
 
 
 v5.3.2
diff --git a/pynamodb/attributes.py b/pynamodb/attributes.py
index d06e3be..c0f182b 100644
--- a/pynamodb/attributes.py
+++ b/pynamodb/attributes.py
@@ -772,7 +772,7 @@ class NumberSetAttribute(Attribute[Set[float]]):
 
 class VersionAttribute(NumberAttribute):
     """
-    A version attribute
+    A number attribute that implements :ref:`optimistic locking <optimistic_locking>`.
     """
     null = True
 
diff --git a/pynamodb/models.py b/pynamodb/models.py
index 3e6f755..16c690b 100644
--- a/pynamodb/models.py
+++ b/pynamodb/models.py
@@ -395,26 +395,35 @@ class Model(AttributeContainer, metaclass=MetaModel):
         """
         return BatchWrite(cls, auto_commit=auto_commit, settings=settings)
 
-    def delete(self, condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default) -> Any:
+    def delete(self, condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default,
+               *, add_version_condition: bool = True) -> Any:
         """
-        Deletes this object from dynamodb
+        Deletes this object from DynamoDB.
 
+        :param add_version_condition: For models which have a :class:`~pynamodb.attributes.VersionAttribute`,
+          specifies whether the item should only be deleted if its current version matches the expected one.
+          Set to `False` for a 'delete anyway' strategy.
         :raises pynamodb.exceptions.DeleteError: If the record can not be deleted
         """
         hk_value, rk_value = self._get_hash_range_key_serialized_values()
+
         version_condition = self._handle_version_attribute()
-        if version_condition is not None:
+        if add_version_condition and version_condition is not None:
             condition &= version_condition
 
         return self._get_connection().delete_item(hk_value, range_key=rk_value, condition=condition, settings=settings)
 
-    def update(self, actions: List[Action], condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default) -> Any:
+    def update(self, actions: List[Action], condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default, *, add_version_condition: bool = True) -> Any:
         """
         Updates an item using the UpdateItem operation.
 
         :param actions: a list of Action updates to apply
         :param condition: an optional Condition on which to update
         :param settings: per-operation settings
+        :param add_version_condition: For models which have a :class:`~pynamodb.attributes.VersionAttribute`,
+          specifies whether only to update if the version matches the model that is currently loaded.
+          Set to `False` for a 'last write wins' strategy.
+          Regardless, the version will always be incremented to prevent "rollbacks" by concurrent :meth:`save` calls.
         :raises ModelInstance.DoesNotExist: if the object to be updated does not exist
         :raises pynamodb.exceptions.UpdateError: if the `condition` is not met
         """
@@ -423,7 +432,7 @@ class Model(AttributeContainer, metaclass=MetaModel):
 
         hk_value, rk_value = self._get_hash_range_key_serialized_values()
         version_condition = self._handle_version_attribute(actions=actions)
-        if version_condition is not None:
+        if add_version_condition and version_condition is not None:
             condition &= version_condition
 
         data = self._get_connection().update_item(hk_value, range_key=rk_value, return_values=ALL_NEW, condition=condition, actions=actions, settings=settings)
@@ -434,11 +443,11 @@ class Model(AttributeContainer, metaclass=MetaModel):
         self.deserialize(item_data)
         return data
 
-    def save(self, condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default) -> Dict[str, Any]:
+    def save(self, condition: Optional[Condition] = None, settings: OperationSettings = OperationSettings.default, *, add_version_condition: bool = True) -> Dict[str, Any]:
         """
         Save this object to dynamodb
         """
-        args, kwargs = self._get_save_args(condition=condition)
+        args, kwargs = self._get_save_args(condition=condition, add_version_condition=add_version_condition)
         kwargs['settings'] = settings
         data = self._get_connection().put_item(*args, **kwargs)
         self.update_local_version_attribute()
@@ -467,11 +476,13 @@ class Model(AttributeContainer, metaclass=MetaModel):
         actions: List[Action],
         condition: Optional[Condition] = None,
         return_values_on_condition_failure: Optional[str] = None,
+        *,
+        add_version_condition: bool = True,
     ) -> Dict[str, Any]:
         hk_value, rk_value = self._get_hash_range_key_serialized_values()
 
         version_condition = self._handle_version_attribute(actions=actions)
-        if version_condition is not None:
+        if add_version_condition and version_condition is not None:
             condition &= version_condition
 
         return self._get_connection().get_operation_kwargs(hk_value, range_key=rk_value, key=KEY, actions=actions, condition=condition, return_values_on_condition_failure=return_values_on_condition_failure)
@@ -480,11 +491,13 @@ class Model(AttributeContainer, metaclass=MetaModel):
         self,
         condition: Optional[Condition] = None,
         return_values_on_condition_failure: Optional[str] = None,
+        *,
+        add_version_condition: bool = True,
     ) -> Dict[str, Any]:
         hk_value, rk_value = self._get_hash_range_key_serialized_values()
 
         version_condition = self._handle_version_attribute()
-        if version_condition is not None:
+        if add_version_condition and version_condition is not None:
             condition &= version_condition
 
         return self._get_connection().get_operation_kwargs(hk_value, range_key=rk_value, key=KEY, condition=condition, return_values_on_condition_failure=return_values_on_condition_failure)
@@ -886,13 +899,16 @@ class Model(AttributeContainer, metaclass=MetaModel):
 
         return schema
 
-    def _get_save_args(self, condition: Optional[Condition] = None) -> Tuple[Iterable[Any], Dict[str, Any]]:
+    def _get_save_args(self, condition: Optional[Condition] = None, *, add_version_condition: bool = True) -> Tuple[Iterable[Any], Dict[str, Any]]:
         """
         Gets the proper *args, **kwargs for saving and retrieving this object
 
         This is used for serializing items to be saved.
 
         :param condition: If set, a condition
+        :param add_version_condition: For models which have a :class:`~pynamodb.attributes.VersionAttribute`,
+          specifies whether the item should only be saved if its current version matches the expected one.
+          Set to `False` for a 'last-write-wins' strategy.
         """
         attribute_values = self.serialize(null_check=True)
         hash_key_attribute = self._hash_key_attribute()
@@ -906,7 +922,7 @@ class Model(AttributeContainer, metaclass=MetaModel):
         if range_key is not None:
             kwargs['range_key'] = range_key
         version_condition = self._handle_version_attribute(attributes=attribute_values)
-        if version_condition is not None:
+        if add_version_condition and version_condition is not None:
             condition &= version_condition
         kwargs['attributes'] = attribute_values
         kwargs['condition'] = condition
diff --git a/pynamodb/transactions.py b/pynamodb/transactions.py
index acc05fc..5b14109 100644
--- a/pynamodb/transactions.py
+++ b/pynamodb/transactions.py
@@ -100,8 +100,11 @@ class TransactWrite(Transaction):
         )
         self._condition_check_items.append(operation_kwargs)
 
-    def delete(self, model: _M, condition: Optional[Condition] = None) -> None:
-        operation_kwargs = model.get_delete_kwargs_from_instance(condition=condition)
+    def delete(self, model: _M, condition: Optional[Condition] = None, *, add_version_condition: bool = True) -> None:
+        operation_kwargs = model.get_delete_kwargs_from_instance(
+            condition=condition,
+            add_version_condition=add_version_condition,
+        )
         self._delete_items.append(operation_kwargs)
 
     def save(self, model: _M, condition: Optional[Condition] = None, return_values: Optional[str] = None) -> None:
@@ -112,11 +115,15 @@ class TransactWrite(Transaction):
         self._put_items.append(operation_kwargs)
         self._models_for_version_attribute_update.append(model)
 
-    def update(self, model: _M, actions: List[Action], condition: Optional[Condition] = None, return_values: Optional[str] = None) -> None:
+    def update(self, model: _M, actions: List[Action], condition: Optional[Condition] = None,
+               return_values: Optional[str] = None,
+               *,
+               add_version_condition: bool = True) -> None:
         operation_kwargs = model.get_update_kwargs_from_instance(
             actions=actions,
             condition=condition,
-            return_values_on_condition_failure=return_values
+            return_values_on_condition_failure=return_values,
+            add_version_condition=add_version_condition,
         )
         self._update_items.append(operation_kwargs)
         self._models_for_version_attribute_update.append(model)


Test patch:
diff --git a/tests/integration/test_transaction_integration.py b/tests/integration/test_transaction_integration.py
index 3d230fb..48ef041 100644
--- a/tests/integration/test_transaction_integration.py
+++ b/tests/integration/test_transaction_integration.py
@@ -323,6 +323,11 @@ def test_transaction_write_with_version_attribute(connection):
     foo3 = Foo(3)
     foo3.save()
 
+    foo42 = Foo(42)
+    foo42.save()
+    foo42_dup = Foo.get(42)
+    foo42_dup.save()  # increment version w/o letting foo4 "know"
+
     with TransactWrite(connection=connection) as transaction:
         transaction.condition_check(Foo, 1, condition=(Foo.bar.exists()))
         transaction.delete(foo2)
@@ -333,6 +338,13 @@ def test_transaction_write_with_version_attribute(connection):
                 Foo.star.set('birdistheword'),
             ]
         )
+        transaction.update(
+            foo42,
+            actions=[
+                Foo.star.set('last write wins'),
+            ],
+            add_version_condition=False,
+        )
 
     assert Foo.get(1).version == 1
     with pytest.raises(DoesNotExist):
@@ -340,6 +352,9 @@ def test_transaction_write_with_version_attribute(connection):
     # Local object's version attribute is updated automatically.
     assert foo3.version == 2
     assert Foo.get(4).version == 1
+    foo42 = Foo.get(42)
+    assert foo42.version == foo42_dup.version + 1 == 3  # ensure version is incremented
+    assert foo42.star == 'last write wins'  # ensure last write wins
 
 
 @pytest.mark.ddblocal
@@ -369,7 +384,8 @@ def test_transaction_write_with_version_attribute_condition_failure(connection):
         with TransactWrite(connection=connection) as transaction:
             transaction.save(Foo(21))
     assert exc_info.value.cause_response_code == TRANSACTION_CANCELLED
-    assert 'ConditionalCheckFailed' in exc_info.value.cause_response_message
+    assert len(exc_info.value.cancellation_reasons) == 1
+    assert exc_info.value.cancellation_reasons[0].code == 'ConditionalCheckFailed'
     assert isinstance(exc_info.value.cause, botocore.exceptions.ClientError)
     assert Foo.Meta.table_name in exc_info.value.cause.MSG_TEMPLATE
 
@@ -382,7 +398,8 @@ def test_transaction_write_with_version_attribute_condition_failure(connection):
                 ]
             )
     assert exc_info.value.cause_response_code == TRANSACTION_CANCELLED
-    assert 'ConditionalCheckFailed' in exc_info.value.cause_response_message
+    assert len(exc_info.value.cancellation_reasons) == 1
+    assert exc_info.value.cancellation_reasons[0].code == 'ConditionalCheckFailed'
     assert Foo.Meta.table_name in exc_info.value.cause.MSG_TEMPLATE
     # Version attribute is not updated on failure.
     assert foo2.version is None
@@ -391,5 +408,6 @@ def test_transaction_write_with_version_attribute_condition_failure(connection):
         with TransactWrite(connection=connection) as transaction:
             transaction.delete(foo2)
     assert exc_info.value.cause_response_code == TRANSACTION_CANCELLED
-    assert 'ConditionalCheckFailed' in exc_info.value.cause_response_message
+    assert len(exc_info.value.cancellation_reasons) == 1
+    assert exc_info.value.cancellation_reasons[0].code == 'ConditionalCheckFailed'
     assert Foo.Meta.table_name in exc_info.value.cause.MSG_TEMPLATE
diff --git a/tests/test_model.py b/tests/test_model.py
index 8d005fa..8e965dc 100644
--- a/tests/test_model.py
+++ b/tests/test_model.py
@@ -722,88 +722,6 @@ class ModelTestCase(TestCase):
             req.return_value = COMPLEX_ITEM_DATA
             item.refresh()
 
-    def test_delete(self):
-        """
-        Model.delete
-        """
-        item = UserModel('foo', 'bar')
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = None
-            item.delete()
-            params = {
-                'Key': {
-                    'user_id': {
-                        'S': 'bar'
-                    },
-                    'user_name': {
-                        'S': 'foo'
-                    }
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'TableName': 'UserModel'
-            }
-            args = req.call_args[0][1]
-            deep_eq(args, params, _assert=True)
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = None
-            item.delete(UserModel.user_id =='bar')
-            params = {
-                'Key': {
-                    'user_id': {
-                        'S': 'bar'
-                    },
-                    'user_name': {
-                        'S': 'foo'
-                    }
-                },
-                'ConditionExpression': '#0 = :0',
-                'ExpressionAttributeNames': {
-                    '#0': 'user_id'
-                },
-                'ExpressionAttributeValues': {
-                    ':0': {
-                        'S': 'bar'
-                    }
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'TableName': 'UserModel'
-            }
-            args = req.call_args[0][1]
-            deep_eq(args, params, _assert=True)
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = None
-            item.delete(UserModel.user_id == 'bar')
-            params = {
-                'Key': {
-                    'user_id': {
-                        'S': 'bar'
-                    },
-                    'user_name': {
-                        'S': 'foo'
-                    }
-                },
-                'ConditionExpression': '#0 = :0',
-                'ExpressionAttributeNames': {
-                    '#0': 'user_id'
-                },
-                'ExpressionAttributeValues': {
-                    ':0': {
-                        'S': 'bar'
-                    }
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'TableName': 'UserModel'
-            }
-            args = req.call_args[0][1]
-            self.assertEqual(args, params)
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = None
-            deep_eq(args, params, _assert=True)
-
     def test_delete_doesnt_do_validation_on_null_attributes(self):
         """
         Model.delete
@@ -3025,58 +2943,6 @@ class ModelTestCase(TestCase):
             self.assert_dict_lists_equal(actual['AttributeDefinitions'],
                                          DOG_TABLE_DATA['Table']['AttributeDefinitions'])
 
-    def test_model_version_attribute_save(self):
-        item = VersionedModel('test_user_name', email='test_user@email.com')
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = {}
-            item.save()
-            args = req.call_args[0][1]
-            params = {
-                'Item': {
-                    'name': {
-                        'S': 'test_user_name'
-                    },
-                    'email': {
-                        'S': 'test_user@email.com'
-                    },
-                    'version': {
-                        'N': '1'
-                    },
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'TableName': 'VersionedModel',
-                'ConditionExpression': 'attribute_not_exists (#0)',
-                'ExpressionAttributeNames': {'#0': 'version'},
-            }
-
-            deep_eq(args, params, _assert=True)
-            item.version = 1
-            item.name = "test_new_username"
-            item.save()
-            args = req.call_args[0][1]
-
-            params = {
-                'Item': {
-                    'name': {
-                        'S': 'test_new_username'
-                    },
-                    'email': {
-                        'S': 'test_user@email.com'
-                    },
-                    'version': {
-                        'N': '2'
-                    },
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'TableName': 'VersionedModel',
-                'ConditionExpression': '#0 = :0',
-                'ExpressionAttributeNames': {'#0': 'version'},
-                'ExpressionAttributeValues': {':0': {'N': '1'}}
-            }
-
-            deep_eq(args, params, _assert=True)
-
     def test_model_version_attribute_save_with_initial_version_zero(self):
         item = VersionedModel('test_user_name', email='test_user@email.com', version=0)
 
@@ -3130,99 +2996,6 @@ class ModelTestCase(TestCase):
 
             deep_eq(args, params, _assert=True)
 
-    def test_version_attribute_increments_on_update(self):
-        item = VersionedModel('test_user_name', email='test_user@email.com')
-
-        with patch(PATCH_METHOD) as req:
-            req.return_value = {
-                ATTRIBUTES: {
-                    'name': {
-                        'S': 'test_user_name'
-                    },
-                    'email': {
-                        'S': 'new@email.com'
-                    },
-                    'version': {
-                        'N': '1'
-                    },
-                }
-            }
-            item.update(actions=[VersionedModel.email.set('new@email.com')])
-            args = req.call_args[0][1]
-            params = {
-                'ConditionExpression': 'attribute_not_exists (#0)',
-                'ExpressionAttributeNames': {
-                    '#0': 'version',
-                    '#1': 'email'
-                },
-                'ExpressionAttributeValues': {
-                    ':0': {
-                        'S': 'new@email.com'
-                    },
-                    ':1': {
-                        'N': '1'
-                    }
-                },
-                'Key': {
-                    'name': {
-                        'S': 'test_user_name'
-                    }
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'ReturnValues': 'ALL_NEW',
-                'TableName': 'VersionedModel',
-                'UpdateExpression': 'SET #1 = :0, #0 = :1'
-            }
-
-            deep_eq(args, params, _assert=True)
-            assert item.version == 1
-
-            req.return_value = {
-                ATTRIBUTES: {
-                    'name': {
-                        'S': 'test_user_name'
-                    },
-                    'email': {
-                        'S': 'newer@email.com'
-                    },
-                    'version': {
-                        'N': '2'
-                    },
-                }
-            }
-
-            item.update(actions=[VersionedModel.email.set('newer@email.com')])
-            args = req.call_args[0][1]
-            params = {
-                'ConditionExpression': '#0 = :0',
-                'ExpressionAttributeNames': {
-                    '#0': 'version',
-                    '#1': 'email'
-                },
-                'ExpressionAttributeValues': {
-                    ':0': {
-                        'N': '1'
-                    },
-                    ':1': {
-                        'S': 'newer@email.com'
-                    },
-                    ':2': {
-                        'N': '1'
-                    }
-                },
-                'Key': {
-                    'name': {
-                        'S': 'test_user_name'
-                    }
-                },
-                'ReturnConsumedCapacity': 'TOTAL',
-                'ReturnValues': 'ALL_NEW',
-                'TableName': 'VersionedModel',
-                'UpdateExpression': 'SET #1 = :1 ADD #0 :2'
-            }
-
-            deep_eq(args, params, _assert=True)
-
 
 class ModelInitTestCase(TestCase):
 
@@ -3383,3 +3156,237 @@ class ModelInitTestCase(TestCase):
         self.assertEqual(Foo._get_connection().table_name, Foo.Meta.table_name)
         self.assertEqual(Bar._get_connection().table_name, Bar.Meta.table_name)
         self.assertEqual(Baz._get_connection().table_name, Baz.Meta.table_name)
+
+
+@pytest.mark.parametrize('add_version_condition', [True, False])
+def test_model_version_attribute_save(add_version_condition: bool) -> None:
+    item = VersionedModel('test_user_name', email='test_user@email.com')
+    with patch(PATCH_METHOD) as req:
+        req.return_value = {}
+        item.save(add_version_condition=add_version_condition)
+        args = req.call_args[0][1]
+        params = {
+            'Item': {
+                'name': {
+                    'S': 'test_user_name'
+                },
+                'email': {
+                    'S': 'test_user@email.com'
+                },
+                'version': {
+                    'N': '1'
+                },
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'TableName': 'VersionedModel',
+        }
+        if add_version_condition:
+            params.update({
+                'ConditionExpression': 'attribute_not_exists (#0)',
+                'ExpressionAttributeNames': {'#0': 'version'},
+            })
+
+        assert args == params
+        deep_eq(args, params, _assert=True)
+        item.version = 1
+        item.name = "test_new_username"
+        item.save(add_version_condition=add_version_condition)
+        args = req.call_args[0][1]
+
+        params = {
+            'Item': {
+                'name': {
+                    'S': 'test_new_username'
+                },
+                'email': {
+                    'S': 'test_user@email.com'
+                },
+                'version': {
+                    'N': '2'
+                },
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'TableName': 'VersionedModel',
+        }
+        if add_version_condition:
+            params.update({
+                'ConditionExpression': '#0 = :0',
+                'ExpressionAttributeNames': {'#0': 'version'},
+                'ExpressionAttributeValues': {':0': {'N': '1'}}
+            })
+
+        assert args == params
+
+
+@pytest.mark.parametrize('add_version_condition', [True, False])
+def test_version_attribute_increments_on_update(add_version_condition: bool) -> None:
+    item = VersionedModel('test_user_name', email='test_user@email.com')
+
+    with patch(PATCH_METHOD) as req:
+        req.return_value = {
+            ATTRIBUTES: {
+                'name': {
+                    'S': 'test_user_name'
+                },
+                'email': {
+                    'S': 'new@email.com'
+                },
+                'version': {
+                    'N': '1'
+                },
+            }
+        }
+        item.update(actions=[VersionedModel.email.set('new@email.com')], add_version_condition=add_version_condition)
+        args = req.call_args[0][1]
+        expected = {
+            'ExpressionAttributeValues': {
+                ':0': {
+                    'S': 'new@email.com'
+                },
+                ':1': {
+                    'N': '1'
+                }
+            },
+            'Key': {
+                'name': {
+                    'S': 'test_user_name'
+                }
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'ReturnValues': 'ALL_NEW',
+            'TableName': 'VersionedModel',
+        }
+        if add_version_condition:
+            expected.update({
+                'ConditionExpression': 'attribute_not_exists (#0)',
+                'ExpressionAttributeNames': {'#0': 'version', '#1': 'email'},
+                'UpdateExpression': 'SET #1 = :0, #0 = :1',
+            })
+        else:
+            expected.update({
+                'ExpressionAttributeNames': {'#0': 'email', '#1': 'version'},
+                'UpdateExpression': 'SET #0 = :0, #1 = :1',
+            })
+
+        assert args == expected
+        assert item.version == 1
+
+    with patch(PATCH_METHOD) as req:
+        req.return_value = {
+            ATTRIBUTES: {
+                'name': {
+                    'S': 'test_user_name'
+                },
+                'email': {
+                    'S': 'newer@email.com'
+                },
+                'version': {
+                    'N': '2'
+                },
+            }
+        }
+
+        item.update(actions=[VersionedModel.email.set('newer@email.com')], add_version_condition=add_version_condition)
+        args = req.call_args[0][1]
+        expected = {
+            'Key': {
+                'name': {
+                    'S': 'test_user_name'
+                }
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'ReturnValues': 'ALL_NEW',
+            'TableName': 'VersionedModel',
+        }
+        if add_version_condition:
+            expected.update({
+                'ConditionExpression': '#0 = :0',
+                'ExpressionAttributeNames': {'#0': 'version', '#1': 'email'},
+                'ExpressionAttributeValues': {':0': {'N': '1'}, ':1': {'S': 'newer@email.com'}, ':2': {'N': '1'}},
+                'UpdateExpression': 'SET #1 = :1 ADD #0 :2',
+            })
+        else:
+            expected.update({
+                'ExpressionAttributeValues': {':0': {'S': 'newer@email.com'}, ':1': {'N': '1'}},
+                'ExpressionAttributeNames': {'#0': 'email', '#1': 'version'},
+                'UpdateExpression': 'SET #0 = :0 ADD #1 :1',
+            })
+
+        assert args == expected
+
+
+@pytest.mark.parametrize('add_version_condition', [True, False])
+def test_delete(add_version_condition: bool) -> None:
+    item = UserModel('foo', 'bar')
+
+    with patch(PATCH_METHOD) as req:
+        req.return_value = None
+        item.delete(add_version_condition=add_version_condition)
+        expected = {
+            'Key': {
+                'user_id': {
+                    'S': 'bar'
+                },
+                'user_name': {
+                    'S': 'foo'
+                }
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'TableName': 'UserModel'
+        }
+        args = req.call_args[0][1]
+        assert args == expected
+
+    with patch(PATCH_METHOD) as req:
+        req.return_value = None
+        item.delete(UserModel.user_id =='bar', add_version_condition=add_version_condition)
+        expected = {
+            'Key': {
+                'user_id': {
+                    'S': 'bar'
+                },
+                'user_name': {
+                    'S': 'foo'
+                }
+            },
+            'ConditionExpression': '#0 = :0',
+            'ExpressionAttributeNames': {
+                '#0': 'user_id'
+            },
+            'ExpressionAttributeValues': {
+                ':0': {
+                    'S': 'bar'
+                }
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'TableName': 'UserModel'
+        }
+        args = req.call_args[0][1]
+        assert args == expected
+
+    with patch(PATCH_METHOD) as req:
+        req.return_value = None
+        item.delete(UserModel.user_id == 'bar', add_version_condition=add_version_condition)
+        expected = {
+            'Key': {
+                'user_id': {
+                    'S': 'bar'
+                },
+                'user_name': {
+                    'S': 'foo'
+                }
+            },
+            'ConditionExpression': '#0 = :0',
+            'ExpressionAttributeNames': {
+                '#0': 'user_id'
+            },
+            'ExpressionAttributeValues': {
+                ':0': {
+                    'S': 'bar'
+                }
+            },
+            'ReturnConsumedCapacity': 'TOTAL',
+            'TableName': 'UserModel'
+        }
+        args = req.call_args[0][1]
+        assert args == expected