From 2e2c99329065e551d5ff8cfe4d5e7177f097412b Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 17:54:15 +0530
Subject: [PATCH 1/7] Add partial model saves

---
 aredis_om/model/model.py | 63 ++++++++++++++++++++++++++++++++++++----
 docs/models.md           | 14 +++++++++
 tests/test_hash_model.py | 49 +++++++++++++++++++++++++++++++
 tests/test_json_model.py | 49 +++++++++++++++++++++++++++++++
 4 files changed, 169 insertions(+), 6 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index a8869b78..8c9cac74 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -13,6 +13,7 @@
     Any,
     Callable,
     Dict,
+    Iterable,
     List,
     Literal,
     Mapping,
@@ -2785,6 +2786,7 @@ async def save(
         pipeline: Optional[Pipeline] = None,
         nx: bool = False,
         xx: bool = False,
+        update_fields: Optional[Iterable[str]] = None,
     ) -> Optional["Model"]:
         """Save the model instance to Redis.
 
@@ -2792,6 +2794,8 @@ async def save(
             pipeline: Optional Redis pipeline for batching operations.
             nx: If True, only save if the key does NOT exist (insert-only).
             xx: If True, only save if the key already exists (update-only).
+            update_fields: Optional iterable of field names to save. When provided,
+                only these fields are written to Redis.
 
         Returns:
             The model instance if saved successfully, None if nx/xx condition
@@ -2802,6 +2806,29 @@ async def save(
         """
         raise NotImplementedError
 
+    def _normalize_update_fields(
+        self, update_fields: Optional[Iterable[str]]
+    ) -> Optional[Set[str]]:
+        """Validate and normalize field names supplied to ``save()``."""
+        if update_fields is None:
+            return None
+        if isinstance(update_fields, str):
+            raise TypeError("update_fields must be an iterable of field names")
+
+        fields = set(update_fields)
+        invalid_fields = fields - set(self.__class__.model_fields)
+        primary_key_name = self.__class__._meta.primary_key.name
+        if primary_key_name in fields:
+            invalid_fields.add(primary_key_name)
+
+        if invalid_fields:
+            field_list = ", ".join(sorted(invalid_fields))
+            raise ValueError(
+                "update_fields contains fields that do not exist on the model or "
+                f"are primary keys: {field_list}"
+            )
+        return fields
+
     async def expire(self, num_seconds: int, pipeline: Optional[Pipeline] = None):
         db = self._get_db(pipeline)
 
@@ -3114,6 +3141,7 @@ async def save(
         nx: bool = False,
         xx: bool = False,
         field_expirations: Optional[Dict[str, int]] = None,
+        update_fields: Optional[Iterable[str]] = None,
     ) -> Optional["Model"]:
         """
         Save the model to Redis.
@@ -3124,6 +3152,8 @@ async def save(
             xx: Only save if the key already exists.
             field_expirations: Dict of {field_name: ttl_seconds} to set field expirations.
                 Overrides any Field(expire=N) defaults. Requires Redis 7.4+.
+            update_fields: Optional iterable of field names to save. When provided,
+                only these fields are written to Redis.
 
         Returns:
             The saved model, or None if nx/xx conditions weren't met.
@@ -3135,12 +3165,18 @@ async def save(
                 "Cannot use nx or xx with pipeline for HashModel. "
                 "Use JsonModel if you need conditional saves with pipelines."
             )
+        if update_fields is not None and (nx or xx):
+            raise ValueError("Cannot combine update_fields with nx or xx")
+
+        normalized_update_fields = self._normalize_update_fields(update_fields)
+        if normalized_update_fields is not None and not normalized_update_fields:
+            return self
 
         self.check()
         db = self._get_db(pipeline)
 
         # Get model data and apply conversions in the correct order
-        document = self.model_dump()
+        document = self.model_dump(include=normalized_update_fields)
         document = convert_datetime_to_timestamp(document)
         # Convert vector fields (list[float]) to bytes before base64 encoding
         document = convert_vector_to_bytes(document, self.__class__.model_fields)
@@ -3298,7 +3334,7 @@ async def update(self, **field_values):
         validate_model_fields(self.__class__, field_values)
         for field, value in field_values.items():
             setattr(self, field, value)
-        await self.save()
+        await self.save(update_fields=field_values)
 
     @classmethod
     def schema_for_fields(cls):
@@ -3524,15 +3560,22 @@ async def save(
         pipeline: Optional[Pipeline] = None,
         nx: bool = False,
         xx: bool = False,
+        update_fields: Optional[Iterable[str]] = None,
     ) -> Optional["Model"]:
         if nx and xx:
             raise ValueError("Cannot specify both nx and xx")
+        if update_fields is not None and (nx or xx):
+            raise ValueError("Cannot combine update_fields with nx or xx")
+
+        normalized_update_fields = self._normalize_update_fields(update_fields)
+        if normalized_update_fields is not None and not normalized_update_fields:
+            return self
 
         self.check()
         db = self._get_db(pipeline)
 
         # Get model data and apply transformations in the correct order
-        data = self.model_dump()
+        data = self.model_dump(include=normalized_update_fields)
         # Convert datetime objects to timestamps for proper indexing
         data = convert_datetime_to_timestamp(data)
         # Convert bytes to base64 strings for safe JSON storage
@@ -3541,11 +3584,18 @@ async def save(
         data = jsonable_encoder(data)
 
         key = self.key()
-        path = Path.root_path()
 
         async def _do_save(conn):
+            if normalized_update_fields is not None:
+                triplets = [
+                    (key, Path(f"$.{field_name}"), data[field_name])
+                    for field_name in sorted(normalized_update_fields)
+                ]
+                await conn.json().mset(triplets)
+                return self
+
             # JSON.SET supports nx and xx natively
-            result = await conn.json().set(key, path, data, nx=nx, xx=xx)
+            result = await conn.json().set(key, Path.root_path(), data, nx=nx, xx=xx)
             # JSON.SET returns None if nx/xx condition not met, "OK" otherwise
             if result is None:
                 return None
@@ -3601,7 +3651,8 @@ async def update(self, **field_values):
             # Set the target field (the last "part" of the nested update
             # field name) to the target value.
             setattr(obj, target_field, value)
-        await self.save()
+        update_fields = {field.split("__", 1)[0] for field in field_values}
+        await self.save(update_fields=update_fields)
 
     @classmethod
     async def get(cls: Type["Model"], pk: Any) -> "Model":
diff --git a/docs/models.md b/docs/models.md
index 45a6e5f2..858f8d67 100644
--- a/docs/models.md
+++ b/docs/models.md
@@ -228,6 +228,20 @@ result = await andrew.save(xx=True)
 
 Returns `None` if the condition was not met, otherwise returns the model.
 
+### Saving Selected Fields
+
+Use `update_fields` to save only selected fields on an existing model:
+
+```python
+andrew.age = 39
+await andrew.save(update_fields=["age"])
+```
+
+This avoids overwriting other fields when multiple processes have loaded and are
+updating the same model. The `update()` method also writes only the fields passed
+to it. Primary-key and unknown field names are rejected, and `update_fields`
+cannot be combined with `nx` or `xx`.
+
 ### Getting a Model by Primary Key
 
 If you have the primary key of a model, you can call the `get()` method:
diff --git a/tests/test_hash_model.py b/tests/test_hash_model.py
index 787750d0..a0b57d59 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1470,6 +1470,55 @@ async def test_save_nx_with_pipeline_raises_error(m):
             await member.save(pipeline=pipe, nx=True)
 
 
+@py_test_mark_asyncio
+async def test_save_update_fields_preserves_concurrent_changes(m):
+    member = m.Member(
+        id=5000,
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        bio="Original bio",
+    )
+    await member.save()
+
+    first_writer = await m.Member.get(member.id)
+    second_writer = await m.Member.get(member.id)
+
+    first_writer.first_name = "Updated first name"
+    await first_writer.save(update_fields=["first_name"])
+
+    second_writer.last_name = "Updated last name"
+    await second_writer.update(last_name="Updated last name")
+
+    saved = await m.Member.get(member.id)
+    assert saved.first_name == "Updated first name"
+    assert saved.last_name == "Updated last name"
+
+
+@py_test_mark_asyncio
+async def test_save_update_fields_validates_field_names(m):
+    member = m.Member(
+        id=5001,
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        bio="Original bio",
+    )
+
+    with pytest.raises(ValueError, match="unknown"):
+        await member.save(update_fields=["unknown"])
+
+    with pytest.raises(ValueError, match="id"):
+        await member.save(update_fields=["id"])
+
+    with pytest.raises(ValueError, match="Cannot combine"):
+        await member.save(update_fields=["first_name"], xx=True)
+
+
 @py_test_mark_asyncio
 async def test_bytes_field_with_binary_data(key_prefix, redis):
     """Test that bytes fields can store arbitrary binary data including non-UTF8 bytes.
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index b439f6ec..ce9e18a9 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -1696,6 +1696,55 @@ async def test_save_nx_with_pipeline(m, address):
     assert fetched2.first_name == "Kim"
 
 
+@py_test_mark_asyncio
+async def test_save_update_fields_preserves_concurrent_changes(m, address):
+    member = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        address=address,
+    )
+    await member.save()
+
+    first_writer = await m.Member.get(member.pk)
+    second_writer = await m.Member.get(member.pk)
+
+    first_writer.first_name = "Updated first name"
+    first_writer.age = 39
+    await first_writer.save(update_fields=["first_name", "age"])
+
+    second_writer.last_name = "Updated last name"
+    await second_writer.update(last_name="Updated last name")
+
+    saved = await m.Member.get(member.pk)
+    assert saved.first_name == "Updated first name"
+    assert saved.last_name == "Updated last name"
+    assert saved.age == 39
+
+
+@py_test_mark_asyncio
+async def test_save_update_fields_validates_field_names(m, address):
+    member = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        address=address,
+    )
+
+    with pytest.raises(ValueError, match="unknown"):
+        await member.save(update_fields=["unknown"])
+
+    with pytest.raises(ValueError, match="pk"):
+        await member.save(update_fields=["pk"])
+
+    with pytest.raises(ValueError, match="Cannot combine"):
+        await member.save(update_fields=["first_name"], xx=True)
+
+
 @py_test_mark_asyncio
 async def test_schema_for_fields_does_not_modify_dict_during_iteration(m):
     """

From 10b1057397acabbd8a5fcca4d5d648ef31afd18e Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 18:31:06 +0530
Subject: [PATCH 2/7] Handle None in partial hash saves

---
 aredis_om/model/model.py | 10 ++++++++--
 tests/test_hash_model.py | 21 +++++++++++++++++++++
 2 files changed, 29 insertions(+), 2 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index 8c9cac74..282730d6 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -3185,8 +3185,14 @@ async def save(
         # Then apply jsonable encoding for other types
         document = jsonable_encoder(document)
 
-        # filter out values which are `None` because they are not valid in a HSET
-        document = {k: v for k, v in document.items() if v is not None}
+        # Redis HSET cannot store None. Preserve the existing full-save behavior of
+        # omitting None values, but encode an explicitly selected None as an empty
+        # string so partial saves can clear optional fields. HashModel.get converts
+        # empty strings back to None for optional fields.
+        if normalized_update_fields is None:
+            document = {k: v for k, v in document.items() if v is not None}
+        else:
+            document = {k: "" if v is None else v for k, v in document.items()}
 
         # Convert boolean values to "1"/"0" for storage efficiency (Redis HSET doesn't support booleans)
         document = {
diff --git a/tests/test_hash_model.py b/tests/test_hash_model.py
index a0b57d59..bda9f77a 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1497,6 +1497,27 @@ async def test_save_update_fields_preserves_concurrent_changes(m):
     assert saved.last_name == "Updated last name"
 
 
+@py_test_mark_asyncio
+async def test_save_update_fields_clears_optional_field(key_prefix, redis):
+    class Member(HashModel, index=True):
+        name: str
+        bio: Optional[str] = None
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew", bio="Original bio")
+    await member.save()
+
+    await member.update(bio=None)
+
+    assert await redis.hget(member.key(), "bio") == ""
+    saved = await Member.get(member.pk)
+    assert saved.bio is None
+    assert saved.name == "Andrew"
+
+
 @py_test_mark_asyncio
 async def test_save_update_fields_validates_field_names(m):
     member = m.Member(

From 9fa2d48203f28825e67e7378357616a86215f8db Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 19:16:58 +0530
Subject: [PATCH 3/7] Fix partial save concurrency edge cases

---
 aredis_om/model/model.py | 91 +++++++++++++++++++++++++++++++---------
 tests/test_hash_model.py | 25 +++++++++++
 tests/test_json_model.py | 51 ++++++++++++++++++++++
 3 files changed, 148 insertions(+), 19 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index 282730d6..8a400e45 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -440,12 +440,9 @@ def convert_empty_strings_to_none(obj, model_fields):
                 field_info.annotation if hasattr(field_info, "annotation") else None
             )
 
-            # Check if the field is Optional (Union[T, None])
-            is_optional = False
-            if hasattr(field_type, "__origin__") and field_type.__origin__ is Union:
-                args = getattr(field_type, "__args__", ())
-                if type(None) in args:
-                    is_optional = True
+            # get_args() supports both Optional[T] and the Python 3.10+
+            # spelling T | None.
+            is_optional = type(None) in get_args(field_type)
 
             if is_optional:
                 result[key] = None
@@ -2040,6 +2037,9 @@ async def update(self, use_transaction=True, **field_values):
         given fields.
         """
         validate_model_fields(self.model, field_values)
+        primary_key_name = self.model._meta.primary_key.name
+        if primary_key_name in field_values:
+            raise ValueError("Cannot update a model's primary key")
         pipeline = await self.model.db().pipeline() if use_transaction else None
 
         # TODO: async for here?
@@ -2048,7 +2048,7 @@ async def update(self, use_transaction=True, **field_values):
                 setattr(model, field, value)
             # TODO: In the non-transaction case, can we do more to detect
             #  failure responses from Redis?
-            await model.save(pipeline=pipeline)
+            await model.save(pipeline=pipeline, update_fields=field_values)
 
         if pipeline:
             # TODO: Response type?
@@ -3337,6 +3337,7 @@ def redisearch_schema(cls):
         return " ".join(schema_parts)
 
     async def update(self, **field_values):
+        self._normalize_update_fields(field_values)
         validate_model_fields(self.__class__, field_values)
         for field, value in field_values.items():
             setattr(self, field, value)
@@ -3561,6 +3562,42 @@ def __init__(self, *args, **kwargs):
             )
         super().__init__(*args, **kwargs)
 
+    async def _save_json_paths(
+        self: "Model",
+        path_values: Mapping[str, Any],
+        pipeline: Optional[Pipeline] = None,
+    ) -> "Model":
+        """Write JSON paths using commands available in redis-py 4.2+."""
+        db = self._get_db(pipeline)
+        key = self.key()
+
+        async def _do_save(conn):
+            # A caller-provided pipeline owns its transaction semantics. Without
+            # one, use MULTI/EXEC so a multi-field partial save stays atomic.
+            if pipeline is not None or len(path_values) == 1:
+                json_conn = conn.json()
+                for json_path, value in path_values.items():
+                    await json_conn.set(key, Path(json_path), value)
+                return self
+
+            async with conn.pipeline(transaction=True) as transaction:
+                json_transaction = transaction.json()
+                for json_path, value in path_values.items():
+                    await json_transaction.set(key, Path(json_path), value)
+                await transaction.execute()
+            return self
+
+        try:
+            return await _do_save(db)
+        except RuntimeError as e:
+            if "Event loop is closed" in str(e):
+                from ..connections import get_redis_connection
+
+                self.__class__._meta.database = get_redis_connection()
+                db = self._get_db(pipeline)
+                return await _do_save(db)
+            raise
+
     async def save(
         self: "Model",
         pipeline: Optional[Pipeline] = None,
@@ -3578,8 +3615,6 @@ async def save(
             return self
 
         self.check()
-        db = self._get_db(pipeline)
-
         # Get model data and apply transformations in the correct order
         data = self.model_dump(include=normalized_update_fields)
         # Convert datetime objects to timestamps for proper indexing
@@ -3589,17 +3624,17 @@ async def save(
         # Apply JSON encoding for complex types (Enums, UUIDs, Sets, etc.)
         data = jsonable_encoder(data)
 
+        if normalized_update_fields is not None:
+            path_values = {
+                f"$.{field_name}": data[field_name]
+                for field_name in sorted(normalized_update_fields)
+            }
+            return await self._save_json_paths(path_values, pipeline=pipeline)
+
+        db = self._get_db(pipeline)
         key = self.key()
 
         async def _do_save(conn):
-            if normalized_update_fields is not None:
-                triplets = [
-                    (key, Path(f"$.{field_name}"), data[field_name])
-                    for field_name in sorted(normalized_update_fields)
-                ]
-                await conn.json().mset(triplets)
-                return self
-
             # JSON.SET supports nx and xx natively
             result = await conn.json().set(key, Path.root_path(), data, nx=nx, xx=xx)
             # JSON.SET returns None if nx/xx condition not met, "OK" otherwise
@@ -3636,6 +3671,11 @@ async def all_pks(cls):  # type: ignore
         )
 
     async def update(self, **field_values):
+        if not field_values:
+            return
+
+        update_fields = {field.split("__", 1)[0] for field in field_values}
+        self._normalize_update_fields(update_fields)
         validate_model_fields(self.__class__, field_values)
         for field, value in field_values.items():
             # Handle the simple update case first, e.g. city="Happy Valley"
@@ -3657,8 +3697,21 @@ async def update(self, **field_values):
             # Set the target field (the last "part" of the nested update
             # field name) to the target value.
             setattr(obj, target_field, value)
-        update_fields = {field.split("__", 1)[0] for field in field_values}
-        await self.save(update_fields=update_fields)
+
+        self.check()
+        data = self.model_dump()
+        data = convert_datetime_to_timestamp(data)
+        data = convert_bytes_to_base64(data)
+        data = jsonable_encoder(data)
+
+        path_values = {}
+        for field in field_values:
+            parts = field.split("__")
+            value = data
+            for part in parts:
+                value = value[part]
+            path_values[f"$.{'.'.join(parts)}"] = value
+        await self._save_json_paths(path_values)
 
     @classmethod
     async def get(cls: Type["Model"], pk: Any) -> "Model":
diff --git a/tests/test_hash_model.py b/tests/test_hash_model.py
index bda9f77a..e416c3bb 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1518,6 +1518,26 @@ class Meta:
     assert saved.name == "Andrew"
 
 
+@py_test_mark_asyncio
+async def test_save_update_fields_clears_pep604_optional_field(key_prefix, redis):
+    class Member(HashModel, index=True):
+        name: str
+        bio: str | None = None
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew", bio="Original bio")
+    await member.save()
+
+    await member.update(bio=None)
+
+    assert await redis.hget(member.key(), "bio") == ""
+    saved = await Member.get(member.pk)
+    assert saved.bio is None
+
+
 @py_test_mark_asyncio
 async def test_save_update_fields_validates_field_names(m):
     member = m.Member(
@@ -1536,6 +1556,11 @@ async def test_save_update_fields_validates_field_names(m):
     with pytest.raises(ValueError, match="id"):
         await member.save(update_fields=["id"])
 
+    original_id = member.id
+    with pytest.raises(ValueError, match="id"):
+        await member.update(id=5002)
+    assert member.id == original_id
+
     with pytest.raises(ValueError, match="Cannot combine"):
         await member.save(update_fields=["first_name"], xx=True)
 
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index ce9e18a9..a733e2ac 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -485,6 +485,29 @@ async def test_update_query(members, m):
     assert all([m.first_name == "Bobby" for m in actual])
 
 
+@py_test_mark_asyncio
+async def test_update_query_preserves_concurrent_changes(members, m):
+    member, _, _ = members
+    original_save = m.Member.save
+    concurrent_change_written = False
+
+    async def save_after_concurrent_change(self, *args, **kwargs):
+        nonlocal concurrent_change_written
+        if not concurrent_change_written:
+            concurrent_change_written = True
+            concurrent_writer = await m.Member.get(self.pk)
+            concurrent_writer.last_name = "Concurrent last name"
+            await original_save(concurrent_writer, update_fields=["last_name"])
+        return await original_save(self, *args, **kwargs)
+
+    with mock.patch.object(m.Member, "save", save_after_concurrent_change):
+        await m.Member.find(m.Member.pk == member.pk).update(first_name="Bobby")
+
+    saved = await m.Member.get(member.pk)
+    assert saved.first_name == "Bobby"
+    assert saved.last_name == "Concurrent last name"
+
+
 @py_test_mark_asyncio
 async def test_exact_match_queries(members, m):
     member1, member2, member3 = members
@@ -1724,6 +1747,29 @@ async def test_save_update_fields_preserves_concurrent_changes(m, address):
     assert saved.age == 39
 
 
+@py_test_mark_asyncio
+async def test_nested_update_preserves_concurrent_sibling_changes(m, address):
+    member = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        address=address,
+    )
+    await member.save()
+
+    first_writer = await m.Member.get(member.pk)
+    second_writer = await m.Member.get(member.pk)
+
+    await first_writer.update(address__city="Seattle")
+    await second_writer.update(address__state="WA")
+
+    saved = await m.Member.get(member.pk)
+    assert saved.address.city == "Seattle"
+    assert saved.address.state == "WA"
+
+
 @py_test_mark_asyncio
 async def test_save_update_fields_validates_field_names(m, address):
     member = m.Member(
@@ -1741,6 +1787,11 @@ async def test_save_update_fields_validates_field_names(m, address):
     with pytest.raises(ValueError, match="pk"):
         await member.save(update_fields=["pk"])
 
+    original_pk = member.pk
+    with pytest.raises(ValueError, match="pk"):
+        await member.update(pk="replacement")
+    assert member.pk == original_pk
+
     with pytest.raises(ValueError, match="Cannot combine"):
         await member.save(update_fields=["first_name"], xx=True)
 

From e12761ca2a5500f40ff7ebb9588b960d2e23905f Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 20:05:48 +0530
Subject: [PATCH 4/7] Harden nested and partial updates

---
 aredis_om/model/model.py | 124 ++++++++++++++++++++++++++-------------
 tests/test_hash_model.py |  52 ++++++++++++++++
 tests/test_json_model.py |  78 ++++++++++++++++++++++++
 3 files changed, 214 insertions(+), 40 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index 8a400e45..dd577d81 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -85,6 +85,13 @@
 _HASH_FIELD_EXPIRATION_MIN_VERSION = (5, 1, 0)
 _HASH_FIELD_EXPIRATION_MIN_SERVER_VERSION = (7, 4)
 _HASH_FIELD_EXPIRATION_SUPPORT_CACHE = weakref.WeakKeyDictionary()
+_HASH_PARTIAL_UPDATE_SCRIPT = """
+if redis.call('EXISTS', KEYS[1]) == 0 then
+    return 0
+end
+redis.call('HSET', KEYS[1], unpack(ARGV))
+return 1
+"""
 
 
 async def supports_hash_field_expiration(conn) -> bool:
@@ -673,18 +680,23 @@ def is_supported_container_type(typ: Optional[type]) -> bool:
 def validate_model_fields(model: Type["RedisModel"], field_values: Dict[str, Any]):
     for field_name in field_values.keys():
         if "__" in field_name:
-            obj = model
+            obj: Any = model
             for sub_field in field_name.split("__"):
-                if not isinstance(obj, ModelMeta) and hasattr(obj, "field"):
-                    obj = getattr(obj, "field").annotation
-
-                if not hasattr(obj, sub_field):
+                model_fields = getattr(obj, "model_fields", {})
+                if sub_field not in model_fields:
                     raise QuerySyntaxError(
                         f"The update path {field_name} contains a field that does not "
                         f"exist on {model.__name__}. The field is: {sub_field}"
                     )
-                obj = getattr(obj, sub_field)
-            return
+                obj = model_fields[sub_field].annotation
+                annotation_args = get_args(obj)
+                if type(None) in annotation_args:
+                    non_none_args = [
+                        arg for arg in annotation_args if arg is not type(None)
+                    ]
+                    if len(non_none_args) == 1:
+                        obj = non_none_args[0]
+            continue
 
         if field_name not in model.model_fields:  # type: ignore
             raise QuerySyntaxError(
@@ -2044,11 +2056,9 @@ async def update(self, use_transaction=True, **field_values):
 
         # TODO: async for here?
         for model in await self.all():
-            for field, value in field_values.items():
-                setattr(model, field, value)
             # TODO: In the non-transaction case, can we do more to detect
             #  failure responses from Redis?
-            await model.save(pipeline=pipeline, update_fields=field_values)
+            await model._update_with_pipeline(field_values, pipeline=pipeline)
 
         if pipeline:
             # TODO: Response type?
@@ -2781,6 +2791,18 @@ async def update(self, **field_values):
         """Update this model instance with the specified key-value pairs."""
         raise NotImplementedError
 
+    async def _update_with_pipeline(
+        self,
+        field_values: Dict[str, Any],
+        pipeline: Optional[Pipeline] = None,
+    ):
+        """Apply and save field updates, optionally using a caller-owned pipeline."""
+        self._normalize_update_fields(field_values)
+        validate_model_fields(self.__class__, field_values)
+        for field, value in field_values.items():
+            setattr(self, field, value)
+        return await self.save(pipeline=pipeline, update_fields=field_values)
+
     async def save(
         self: "Model",
         pipeline: Optional[Pipeline] = None,
@@ -2816,16 +2838,25 @@ def _normalize_update_fields(
             raise TypeError("update_fields must be an iterable of field names")
 
         fields = set(update_fields)
-        invalid_fields = fields - set(self.__class__.model_fields)
+        model_fields = self.__class__.model_fields
+        invalid_fields = fields - set(model_fields)
         primary_key_name = self.__class__._meta.primary_key.name
         if primary_key_name in fields:
             invalid_fields.add(primary_key_name)
+        excluded_fields = {
+            field_name
+            for field_name in fields
+            if field_name in model_fields
+            and getattr(model_fields[field_name], "exclude", False) is True
+        }
+        invalid_fields.update(excluded_fields)
 
         if invalid_fields:
             field_list = ", ".join(sorted(invalid_fields))
             raise ValueError(
                 "update_fields contains fields that do not exist on the model or "
-                f"are primary keys: {field_list}"
+                "are primary keys or excluded from serialization: "
+                f"{field_list}"
             )
         return fields
 
@@ -3232,7 +3263,15 @@ async def _do_save(conn):
                             if current_ttls[i] > 0:  # Has a TTL
                                 preserved_ttls[field_name] = current_ttls[i]
 
-            await conn.hset(key, mapping=document)
+            if normalized_update_fields is None:
+                await conn.hset(key, mapping=document)
+            else:
+                hset_args = [item for pair in document.items() for item in pair]
+                result = await conn.eval(
+                    _HASH_PARTIAL_UPDATE_SCRIPT, 1, key, *hset_args
+                )
+                if not is_pipeline and result == 0:
+                    return None
 
             # Apply field expirations after HSET (requires Redis 7.4+)
             # When using pipelines, we can still apply default expirations but
@@ -3337,11 +3376,7 @@ def redisearch_schema(cls):
         return " ".join(schema_parts)
 
     async def update(self, **field_values):
-        self._normalize_update_fields(field_values)
-        validate_model_fields(self.__class__, field_values)
-        for field, value in field_values.items():
-            setattr(self, field, value)
-        await self.save(update_fields=field_values)
+        await self._update_with_pipeline(field_values)
 
     @classmethod
     def schema_for_fields(cls):
@@ -3610,13 +3645,25 @@ async def save(
         if update_fields is not None and (nx or xx):
             raise ValueError("Cannot combine update_fields with nx or xx")
 
-        normalized_update_fields = self._normalize_update_fields(update_fields)
-        if normalized_update_fields is not None and not normalized_update_fields:
-            return self
+        normalized_update_fields: Optional[Set[str]] = None
+        update_field_roots: Optional[Set[str]] = None
+        if update_fields is not None:
+            if isinstance(update_fields, str):
+                raise TypeError("update_fields must be an iterable of field names")
+            normalized_update_fields = set(update_fields)
+            if not normalized_update_fields:
+                return self
+            update_field_roots = {
+                field_name.split("__", 1)[0] for field_name in normalized_update_fields
+            }
+            self._normalize_update_fields(update_field_roots)
+            validate_model_fields(
+                self.__class__, dict.fromkeys(normalized_update_fields)
+            )
 
         self.check()
         # Get model data and apply transformations in the correct order
-        data = self.model_dump(include=normalized_update_fields)
+        data = self.model_dump(include=update_field_roots)
         # Convert datetime objects to timestamps for proper indexing
         data = convert_datetime_to_timestamp(data)
         # Convert bytes to base64 strings for safe JSON storage
@@ -3625,10 +3672,13 @@ async def save(
         data = jsonable_encoder(data)
 
         if normalized_update_fields is not None:
-            path_values = {
-                f"$.{field_name}": data[field_name]
-                for field_name in sorted(normalized_update_fields)
-            }
+            path_values = {}
+            for field_name in sorted(normalized_update_fields):
+                parts = field_name.split("__")
+                value = data
+                for part in parts:
+                    value = value[part]
+                path_values[f"$.{'.'.join(parts)}"] = value
             return await self._save_json_paths(path_values, pipeline=pipeline)
 
         db = self._get_db(pipeline)
@@ -3670,7 +3720,11 @@ async def all_pks(cls):  # type: ignore
             async for key in cls.db().scan_iter(f"{key_prefix}*", _type="ReJSON-RL")
         )
 
-    async def update(self, **field_values):
+    async def _update_with_pipeline(
+        self,
+        field_values: Dict[str, Any],
+        pipeline: Optional[Pipeline] = None,
+    ):
         if not field_values:
             return
 
@@ -3698,20 +3752,10 @@ async def update(self, **field_values):
             # field name) to the target value.
             setattr(obj, target_field, value)
 
-        self.check()
-        data = self.model_dump()
-        data = convert_datetime_to_timestamp(data)
-        data = convert_bytes_to_base64(data)
-        data = jsonable_encoder(data)
+        return await self.save(pipeline=pipeline, update_fields=field_values)
 
-        path_values = {}
-        for field in field_values:
-            parts = field.split("__")
-            value = data
-            for part in parts:
-                value = value[part]
-            path_values[f"$.{'.'.join(parts)}"] = value
-        await self._save_json_paths(path_values)
+    async def update(self, **field_values):
+        await self._update_with_pipeline(field_values)
 
     @classmethod
     async def get(cls: Type["Model"], pk: Any) -> "Model":
diff --git a/tests/test_hash_model.py b/tests/test_hash_model.py
index e416c3bb..fecc928c 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1497,6 +1497,58 @@ async def test_save_update_fields_preserves_concurrent_changes(m):
     assert saved.last_name == "Updated last name"
 
 
+@py_test_mark_asyncio
+async def test_update_query_uses_partial_hash_save(members, m):
+    member, _, _ = members
+
+    await m.Member.find(m.Member.id == member.id).update(first_name="Bobby")
+
+    saved = await m.Member.get(member.id)
+    assert saved.first_name == "Bobby"
+
+
+@py_test_mark_asyncio
+async def test_save_update_fields_does_not_recreate_deleted_model(m):
+    member = m.Member(
+        id=5003,
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        bio="Original bio",
+    )
+    await member.save()
+    await m.Member.db().delete(member.key())
+
+    member.first_name = "Updated first name"
+    result = await member.save(update_fields=["first_name"])
+
+    assert result is None
+    assert not await m.Member.db().exists(member.key())
+
+
+@py_test_mark_asyncio
+async def test_save_update_fields_rejects_excluded_fields(key_prefix, redis):
+    class Member(HashModel, index=True):
+        name: str
+        transient: str = Field(default="hidden", exclude=True)
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew")
+    await member.save()
+
+    with pytest.raises(ValueError, match="transient"):
+        await member.save(update_fields=["transient"])
+
+    with pytest.raises(ValueError, match="transient"):
+        await member.update(transient="changed")
+    assert member.transient == "hidden"
+
+
 @py_test_mark_asyncio
 async def test_save_update_fields_clears_optional_field(key_prefix, redis):
     class Member(HashModel, index=True):
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index a733e2ac..9a14f88f 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -22,6 +22,7 @@
     Migrator,
     NotFoundError,
     QueryNotSupportedError,
+    QuerySyntaxError,
     RedisModel,
     RedisModelError,
     VectorFieldOptions,
@@ -508,6 +509,19 @@ async def save_after_concurrent_change(self, *args, **kwargs):
     assert saved.last_name == "Concurrent last name"
 
 
+@py_test_mark_asyncio
+async def test_update_query_supports_nested_fields(members, m):
+    member, _, _ = members
+
+    await m.Member.find(m.Member.pk == member.pk).update(
+        address__city="Seattle", first_name="Bobby"
+    )
+
+    saved = await m.Member.get(member.pk)
+    assert saved.address.city == "Seattle"
+    assert saved.first_name == "Bobby"
+
+
 @py_test_mark_asyncio
 async def test_exact_match_queries(members, m):
     member1, member2, member3 = members
@@ -1770,6 +1784,70 @@ async def test_nested_update_preserves_concurrent_sibling_changes(m, address):
     assert saved.address.state == "WA"
 
 
+@py_test_mark_asyncio
+async def test_nested_update_validates_every_path_before_mutation(m, address):
+    member = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        address=address,
+    )
+    await member.save()
+
+    with pytest.raises(QuerySyntaxError, match="missing"):
+        await member.update(address__city="Seattle", address__missing="value")
+
+    assert member.address.city == "Portland"
+    saved = await m.Member.get(member.pk)
+    assert saved.address.city == "Portland"
+
+
+@py_test_mark_asyncio
+async def test_nested_update_supports_optional_embedded_models(key_prefix, redis):
+    class Address(EmbeddedJsonModel):
+        city: str
+
+    class Member(JsonModel, index=True):
+        name: str
+        address: Optional[Address] = None
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew", address=Address(city="Portland"))
+    await member.save()
+
+    await member.update(address__city="Seattle")
+
+    saved = await Member.get(member.pk)
+    assert saved.address is not None
+    assert saved.address.city == "Seattle"
+
+
+@py_test_mark_asyncio
+async def test_save_update_fields_rejects_excluded_fields(key_prefix, redis):
+    class Member(JsonModel, index=True):
+        name: str
+        transient: str = Field(default="hidden", exclude=True)
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew")
+    await member.save()
+
+    with pytest.raises(ValueError, match="transient"):
+        await member.save(update_fields=["transient"])
+
+    with pytest.raises(ValueError, match="transient"):
+        await member.update(transient="changed")
+    assert member.transient == "hidden"
+
+
 @py_test_mark_asyncio
 async def test_save_update_fields_validates_field_names(m, address):
     member = m.Member(

From bf945ca4bb8ab41549054b40fcc2dc6855255376 Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 20:28:27 +0530
Subject: [PATCH 5/7] Handle deleted JSON partial-save targets

---
 aredis_om/model/model.py | 43 +++++++++++++++++++++++-----------------
 docs/models.md           |  3 +++
 tests/test_json_model.py | 35 ++++++++++++++++++++++++++++++++
 3 files changed, 63 insertions(+), 18 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index dd577d81..ae480cf4 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -92,6 +92,15 @@
 redis.call('HSET', KEYS[1], unpack(ARGV))
 return 1
 """
+_JSON_PARTIAL_UPDATE_SCRIPT = """
+if redis.call('EXISTS', KEYS[1]) == 0 then
+    return 0
+end
+for i = 1, #ARGV, 2 do
+    redis.call('JSON.SET', KEYS[1], ARGV[i], ARGV[i + 1])
+end
+return 1
+"""
 
 
 async def supports_hash_field_expiration(conn) -> bool:
@@ -2820,8 +2829,8 @@ async def save(
                 only these fields are written to Redis.
 
         Returns:
-            The model instance if saved successfully, None if nx/xx condition
-            was not met.
+            The model instance if saved successfully, or None if an nx/xx condition
+            was not met or a partial-save target no longer exists.
 
         Raises:
             ValueError: If both nx and xx are True.
@@ -3187,7 +3196,8 @@ async def save(
                 only these fields are written to Redis.
 
         Returns:
-            The saved model, or None if nx/xx conditions weren't met.
+            The saved model, or None if nx/xx conditions weren't met or a
+            partial-save target no longer exists.
         """
         if nx and xx:
             raise ValueError("Cannot specify both nx and xx")
@@ -3601,25 +3611,22 @@ async def _save_json_paths(
         self: "Model",
         path_values: Mapping[str, Any],
         pipeline: Optional[Pipeline] = None,
-    ) -> "Model":
-        """Write JSON paths using commands available in redis-py 4.2+."""
+    ) -> Optional["Model"]:
+        """Atomically update JSON paths only when the model still exists."""
         db = self._get_db(pipeline)
         key = self.key()
+        json_set_args = [
+            item
+            for path, value in path_values.items()
+            for item in (path, json.dumps(value))
+        ]
 
         async def _do_save(conn):
-            # A caller-provided pipeline owns its transaction semantics. Without
-            # one, use MULTI/EXEC so a multi-field partial save stays atomic.
-            if pipeline is not None or len(path_values) == 1:
-                json_conn = conn.json()
-                for json_path, value in path_values.items():
-                    await json_conn.set(key, Path(json_path), value)
-                return self
-
-            async with conn.pipeline(transaction=True) as transaction:
-                json_transaction = transaction.json()
-                for json_path, value in path_values.items():
-                    await json_transaction.set(key, Path(json_path), value)
-                await transaction.execute()
+            result = await conn.eval(
+                _JSON_PARTIAL_UPDATE_SCRIPT, 1, key, *json_set_args
+            )
+            if pipeline is None and result == 0:
+                return None
             return self
 
         try:
diff --git a/docs/models.md b/docs/models.md
index 858f8d67..3c916e19 100644
--- a/docs/models.md
+++ b/docs/models.md
@@ -242,6 +242,9 @@ updating the same model. The `update()` method also writes only the fields passe
 to it. Primary-key and unknown field names are rejected, and `update_fields`
 cannot be combined with `nx` or `xx`.
 
+If the model is deleted before a partial save is applied, `save()` returns `None`
+and does not recreate an incomplete record.
+
 ### Getting a Model by Primary Key
 
 If you have the primary key of a model, you can call the `get()` method:
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index 9a14f88f..8f003194 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -522,6 +522,21 @@ async def test_update_query_supports_nested_fields(members, m):
     assert saved.first_name == "Bobby"
 
 
+@py_test_mark_asyncio
+async def test_update_query_skips_concurrently_deleted_model(members, m):
+    member, _, _ = members
+    original_update = m.Member._update_with_pipeline
+
+    async def update_after_delete(self, field_values, pipeline=None):
+        await self.__class__.db().delete(self.key())
+        return await original_update(self, field_values, pipeline=pipeline)
+
+    with mock.patch.object(m.Member, "_update_with_pipeline", update_after_delete):
+        await m.Member.find(m.Member.pk == member.pk).update(first_name="Bobby")
+
+    assert not await m.Member.db().exists(member.key())
+
+
 @py_test_mark_asyncio
 async def test_exact_match_queries(members, m):
     member1, member2, member3 = members
@@ -1761,6 +1776,26 @@ async def test_save_update_fields_preserves_concurrent_changes(m, address):
     assert saved.age == 39
 
 
+@py_test_mark_asyncio
+async def test_save_update_fields_skips_deleted_model(m, address):
+    member = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        address=address,
+    )
+    await member.save()
+    await m.Member.db().delete(member.key())
+
+    member.first_name = "Bobby"
+    result = await member.save(update_fields=["first_name"])
+
+    assert result is None
+    assert not await m.Member.db().exists(member.key())
+
+
 @py_test_mark_asyncio
 async def test_nested_update_preserves_concurrent_sibling_changes(m, address):
     member = m.Member(

From c54e915bb1e8eb0cde3bbce023aaca99e2baf129 Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Fri, 4 Sep 2026 20:53:45 +0530
Subject: [PATCH 6/7] Reject excluded nested update fields

---
 aredis_om/model/model.py | 13 ++++++++++++-
 tests/test_json_model.py | 35 +++++++++++++++++++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index ae480cf4..2f16c4d7 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -697,7 +697,13 @@ def validate_model_fields(model: Type["RedisModel"], field_values: Dict[str, Any
                         f"The update path {field_name} contains a field that does not "
                         f"exist on {model.__name__}. The field is: {sub_field}"
                     )
-                obj = model_fields[sub_field].annotation
+                field_info = model_fields[sub_field]
+                if getattr(field_info, "exclude", False) is True:
+                    raise ValueError(
+                        "update_fields contains a field excluded from serialization: "
+                        f"{field_name}"
+                    )
+                obj = field_info.annotation
                 annotation_args = get_args(obj)
                 if type(None) in annotation_args:
                     non_none_args = [
@@ -711,6 +717,11 @@ def validate_model_fields(model: Type["RedisModel"], field_values: Dict[str, Any
             raise QuerySyntaxError(
                 f"The field {field_name} does not exist on the model {model.__name__}"
             )
+        if getattr(model.model_fields[field_name], "exclude", False) is True:
+            raise ValueError(
+                "update_fields contains a field excluded from serialization: "
+                f"{field_name}"
+            )
 
 
 def decode_redis_value(
diff --git a/tests/test_json_model.py b/tests/test_json_model.py
index 8f003194..c1897821 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -1882,6 +1882,41 @@ class Meta:
         await member.update(transient="changed")
     assert member.transient == "hidden"
 
+    with pytest.raises(ValueError, match="transient"):
+        await Member.find().update(transient="changed")
+
+
+@py_test_mark_asyncio
+async def test_nested_update_rejects_excluded_fields_before_mutation(key_prefix, redis):
+    class Address(EmbeddedJsonModel):
+        city: str
+        transient: str = Field(default="hidden", exclude=True)
+
+    class Member(JsonModel, index=True):
+        name: str
+        address: Address
+
+        class Meta:
+            global_key_prefix = key_prefix
+            database = redis
+
+    member = Member(name="Andrew", address=Address(city="Portland"))
+    await member.save()
+
+    with pytest.raises(ValueError, match="address__transient"):
+        await member.update(address__transient="changed")
+    assert member.address.transient == "hidden"
+
+    member.address.transient = "changed"
+    with pytest.raises(ValueError, match="address__transient"):
+        await member.save(update_fields=["address__transient"])
+
+
+@py_test_mark_asyncio
+async def test_update_query_rejects_embedded_primary_key(m):
+    with pytest.raises(ValueError, match="address__pk"):
+        await m.Member.find().update(address__pk="replacement")
+
 
 @py_test_mark_asyncio
 async def test_save_update_fields_validates_field_names(m, address):

From f4f0caf03c130c1ba9dee859e10da33842713838 Mon Sep 17 00:00:00 2001
From: Hari Om Tiwari <hariom.codes@gmail.com>
Date: Sat, 5 Sep 2026 18:05:47 +0530
Subject: [PATCH 7/7] Decode partial hash nulls in search results

---
 aredis_om/model/model.py |  4 ++-
 tests/test_hash_model.py | 54 ++++++++++++++++++++++------------------
 2 files changed, 33 insertions(+), 25 deletions(-)

diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index 2f16c4d7..6f57e127 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -2818,7 +2818,6 @@ async def _update_with_pipeline(
     ):
         """Apply and save field updates, optionally using a caller-owned pipeline."""
         self._normalize_update_fields(field_values)
-        validate_model_fields(self.__class__, field_values)
         for field, value in field_values.items():
             setattr(self, field, value)
         return await self.save(pipeline=pipeline, update_fields=field_values)
@@ -3002,6 +3001,9 @@ def to_string(s):
                 json_fields = convert_base64_to_bytes(json_fields, cls.model_fields)
                 doc = cls(**json_fields)
             else:
+                # Match HashModel.get(): decode explicit nulls before validation
+                # and type conversions, including for required nullable fields.
+                fields = convert_empty_strings_to_none(fields, cls.model_fields)
                 # Convert timestamps back to datetime objects
                 fields = convert_timestamp_to_datetime(fields, cls.model_fields)
                 # Convert base64 strings back to bytes for bytes fields
diff --git a/tests/test_hash_model.py b/tests/test_hash_model.py
index fecc928c..555c75b1 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1550,16 +1550,38 @@ class Meta:
 
 
 @py_test_mark_asyncio
-async def test_save_update_fields_clears_optional_field(key_prefix, redis):
+@pytest.mark.parametrize(
+    "annotation,default,initial",
+    [
+        pytest.param(Optional[str], ..., "Original bio", id="required-string"),
+        pytest.param(str | None, None, "Original bio", id="none-default-string"),
+        pytest.param(Optional[str], "fallback", "Original bio", id="default-string"),
+        pytest.param(Optional[int], ..., 42, id="required-integer"),
+        pytest.param(int | None, None, 42, id="none-default-integer"),
+        pytest.param(Optional[int], 7, 42, id="default-integer"),
+        pytest.param(Optional[bool], None, True, id="boolean"),
+        pytest.param(
+            Optional[datetime.datetime],
+            None,
+            datetime.datetime(2026, 1, 1),
+            id="datetime",
+        ),
+    ],
+)
+async def test_save_update_fields_clears_optional_field(
+    key_prefix, redis, annotation, default, initial
+):
     class Member(HashModel, index=True):
-        name: str
-        bio: Optional[str] = None
+        name: str = Field(index=True)
+        # Only name is indexed: numeric null storage needs a separate policy.
+        bio: annotation = default
 
         class Meta:
             global_key_prefix = key_prefix
             database = redis
 
-    member = Member(name="Andrew", bio="Original bio")
+    await Migrator(conn=redis).run()
+    member = Member(name="Andrew", bio=initial)
     await member.save()
 
     await member.update(bio=None)
@@ -1568,26 +1590,10 @@ class Meta:
     saved = await Member.get(member.pk)
     assert saved.bio is None
     assert saved.name == "Andrew"
-
-
-@py_test_mark_asyncio
-async def test_save_update_fields_clears_pep604_optional_field(key_prefix, redis):
-    class Member(HashModel, index=True):
-        name: str
-        bio: str | None = None
-
-        class Meta:
-            global_key_prefix = key_prefix
-            database = redis
-
-    member = Member(name="Andrew", bio="Original bio")
-    await member.save()
-
-    await member.update(bio=None)
-
-    assert await redis.hget(member.key(), "bio") == ""
-    saved = await Member.get(member.pk)
-    assert saved.bio is None
+    found = await Member.find(Member.name == "Andrew").all()
+    assert len(found) == 1
+    assert found[0].pk == member.pk
+    assert found[0].bio is None
 
 
 @py_test_mark_asyncio