diff --git a/aredis_om/model/model.py b/aredis_om/model/model.py
index a8869b78..6f57e127 100644
--- a/aredis_om/model/model.py
+++ b/aredis_om/model/model.py
@@ -13,6 +13,7 @@
     Any,
     Callable,
     Dict,
+    Iterable,
     List,
     Literal,
     Mapping,
@@ -84,6 +85,22 @@
 _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
+"""
+_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:
@@ -439,12 +456,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
@@ -675,23 +689,39 @@ 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
+                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 = [
+                        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(
                 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(
@@ -2039,15 +2069,16 @@ 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?
         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)
+            await model._update_with_pipeline(field_values, pipeline=pipeline)
 
         if pipeline:
             # TODO: Response type?
@@ -2780,11 +2811,23 @@ 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)
+        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,
         nx: bool = False,
         xx: bool = False,
+        update_fields: Optional[Iterable[str]] = None,
     ) -> Optional["Model"]:
         """Save the model instance to Redis.
 
@@ -2792,16 +2835,50 @@ 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
-            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.
         """
         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)
+        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 "
+                "are primary keys or excluded from serialization: "
+                f"{field_list}"
+            )
+        return fields
+
     async def expire(self, num_seconds: int, pipeline: Optional[Pipeline] = None):
         db = self._get_db(pipeline)
 
@@ -2924,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
@@ -3114,6 +3194,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,9 +3205,12 @@ 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.
+            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")
@@ -3135,12 +3219,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)
@@ -3149,8 +3239,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 = {
@@ -3190,7 +3286,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
@@ -3295,10 +3399,7 @@ def redisearch_schema(cls):
         return " ".join(schema_parts)
 
     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._update_with_pipeline(field_values)
 
     @classmethod
     def schema_for_fields(cls):
@@ -3519,20 +3620,70 @@ 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,
+    ) -> 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):
+            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:
+            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,
         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: 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()
-        db = self._get_db(pipeline)
-
         # Get model data and apply transformations in the correct order
-        data = self.model_dump()
+        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
@@ -3540,12 +3691,22 @@ 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 = {}
+            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)
         key = self.key()
-        path = Path.root_path()
 
         async def _do_save(conn):
             # 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
@@ -3579,7 +3740,16 @@ 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
+
+        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"
@@ -3601,7 +3771,11 @@ 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()
+
+        return await self.save(pipeline=pipeline, update_fields=field_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/docs/models.md b/docs/models.md
index 45a6e5f2..3c916e19 100644
--- a/docs/models.md
+++ b/docs/models.md
@@ -228,6 +228,23 @@ 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`.
+
+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_hash_model.py b/tests/test_hash_model.py
index 787750d0..555c75b1 100644
--- a/tests/test_hash_model.py
+++ b/tests/test_hash_model.py
@@ -1470,6 +1470,159 @@ 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_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
+@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 = 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
+
+    await Migrator(conn=redis).run()
+    member = Member(name="Andrew", bio=initial)
+    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"
+    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
+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"])
+
+    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)
+
+
 @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..c1897821 100644
--- a/tests/test_json_model.py
+++ b/tests/test_json_model.py
@@ -22,6 +22,7 @@
     Migrator,
     NotFoundError,
     QueryNotSupportedError,
+    QuerySyntaxError,
     RedisModel,
     RedisModelError,
     VectorFieldOptions,
@@ -485,6 +486,57 @@ 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_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_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
@@ -1696,6 +1748,202 @@ 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_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(
+        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_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"
+
+    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):
+    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"])
+
+    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)
+
+
 @py_test_mark_asyncio
 async def test_schema_for_fields_does_not_modify_dict_during_iteration(m):
     """