fix: prevent update() from overwriting unset fields with None (litestar-org#560) (litestar-org#563)

Added `was_attribute_set()` helper function that uses SQLAlchemy's
instance state inspection to check which attributes were actually
modified/set on the input instance. The `update()` method now only copies
attributes that have been explicitly set by the user.

Author: cofin

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: unasyncd
         additional_dependencies: ["ruff"]
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.13.1"
+    rev: "v0.14.0"
     hooks:
       # Run the linter.
       - id: ruff

advanced_alchemy/repository/_async.py

@@ -52,6 +52,7 @@
     compare_values,
     get_abstract_loader_options,
     get_instrumented_attr,
+    was_attribute_set,
 )
 from advanced_alchemy.repository.typing import MISSING, ModelT, OrderingPair, T
 from advanced_alchemy.service.typing import schema_dump
@@ -1502,6 +1503,10 @@ async def update(
                             # This prevents overwriting columns that should use their defaults
                             if new_field_value is None and column_has_defaults(column):
                                 continue
+                            # Only copy attributes that were explicitly set on the input instance
+                            # This prevents overwriting existing values with uninitialized None values
+                            if not was_attribute_set(data, mapper, field_name):
+                                continue
                             existing_field_value = getattr(existing_instance, field_name, MISSING)
                             if existing_field_value is not MISSING and not compare_values(
                                 existing_field_value, new_field_value

advanced_alchemy/repository/_sync.py

@@ -53,6 +53,7 @@
     compare_values,
     get_abstract_loader_options,
     get_instrumented_attr,
+    was_attribute_set,
 )
 from advanced_alchemy.repository.typing import MISSING, ModelT, OrderingPair, T
 from advanced_alchemy.service.typing import schema_dump
@@ -1503,6 +1504,10 @@ def update(
                             # This prevents overwriting columns that should use their defaults
                             if new_field_value is None and column_has_defaults(column):
                                 continue
+                            # Only copy attributes that were explicitly set on the input instance
+                            # This prevents overwriting existing values with uninitialized None values
+                            if not was_attribute_set(data, mapper, field_name):
+                                continue
                             existing_field_value = getattr(existing_instance, field_name, MISSING)
                             if existing_field_value is not MISSING and not compare_values(
                                 existing_field_value, new_field_value

advanced_alchemy/repository/_util.py

@@ -371,6 +371,42 @@ def column_has_defaults(column: Any) -> bool:
     )
 
 
+def was_attribute_set(instance: Any, mapper: Any, attr_name: str) -> bool:
+    """Check if an attribute was explicitly set on a model instance.
+
+    This function distinguishes between attributes that were explicitly set
+    (even to None) versus attributes that are simply uninitialized and defaulting
+    to None. This is crucial for partial updates where only modified fields
+    should be copied.
+
+    Args:
+        instance: The model instance to check.
+        mapper: The SQLAlchemy mapper/inspector for the instance.
+        attr_name: The name of the attribute to check.
+
+    Returns:
+        bool: True if the attribute was explicitly set, False if uninitialized.
+    """
+    try:
+        # Get the attribute state
+        attr_state = mapper.attrs.get(attr_name)
+        if attr_state is None:
+            return False
+
+        # Check if the attribute has history (was modified)
+        # For a new transient instance, modified attributes will have history
+        history = attr_state.history
+        if history.has_changes():
+            return True
+
+        # For attributes with no history, check if they're in the instance dict
+        # This handles the case where an attribute was set during __init__
+        return hasattr(instance, "__dict__") and attr_name in instance.__dict__
+    except (AttributeError, KeyError):  # pragma: no cover
+        # If we can't determine, assume it was set to be safe
+        return True
+
+
 def compare_values(existing_value: Any, new_value: Any) -> bool:
     """Safely compare two values, handling numpy arrays and other special types.
 

tests/integration/test_repository.py

@@ -734,12 +734,10 @@ async def test_repo_update_many_non_returning_backend_refresh(
 
     # Create multiple authors
     authors = await maybe_async(
-        author_repo.create_many(
-            [
-                {"name": "Author A", "dob": datetime.date(1990, 1, 1)},
-                {"name": "Author B", "dob": datetime.date(1991, 2, 2)},
-            ]
-        )
+        author_repo.create_many([
+            {"name": "Author A", "dob": datetime.date(1990, 1, 1)},
+            {"name": "Author B", "dob": datetime.date(1991, 2, 2)},
+        ])
     )
 
     # Prepare update data with partial changes
@@ -775,13 +773,11 @@ async def test_service_mixed_input_types_update_many(
 
     # Create multiple authors
     authors = await maybe_async(
-        author_service.create_many(
-            [
-                {"name": "Author 1", "dob": datetime.date(1990, 1, 1)},
-                {"name": "Author 2", "dob": datetime.date(1991, 2, 2)},
-                {"name": "Author 3", "dob": datetime.date(1992, 3, 3)},
-            ]
-        )
+        author_service.create_many([
+            {"name": "Author 1", "dob": datetime.date(1990, 1, 1)},
+            {"name": "Author 2", "dob": datetime.date(1991, 2, 2)},
+            {"name": "Author 3", "dob": datetime.date(1992, 3, 3)},
+        ])
     )
 
     # Get ID type from model for dynamic schema creation
@@ -831,3 +827,73 @@ class AuthorUpdateMsgspec(msgspec.Struct):  # type: ignore[name-defined,misc]
         assert author.dob is not None
         assert author.created_at is not None
         assert author.updated_at is not None
+
+
+async def test_repo_update_with_model_instance_partial_fields_github_560(
+    seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]",
+) -> None:
+    """Test repository update with model instances for partial updates (GitHub Issue #560).
+
+    This test verifies that when updating with a model instance where only some fields
+    are explicitly set, the unset fields do not overwrite existing data with None.
+    """
+    _session, models = seeded_test_session_async
+    author_model = models["author"]
+    author_repo = get_repository_from_session(seeded_test_session_async, "author")
+
+    # Create an author with all fields populated
+    author = await maybe_async(author_repo.create({"name": "Original Name", "dob": datetime.date(1990, 1, 1)}))
+    original_dob = author.dob
+    original_id = author.id
+
+    # Create a partial update using a model instance with only id and name set
+    # This mimics the pattern: Author(id=1, name="Updated Name")
+    # SQLAlchemy initializes 'dob' to None, but it wasn't explicitly set by the user
+    partial_update = author_model(id=original_id, name="Updated Name")
+
+    # Update via repository - should only update name, leave dob unchanged
+    updated_author = await maybe_async(author_repo.update(partial_update))
+
+    # Verify: name was updated, but dob remains unchanged (not overwritten with None)
+    assert updated_author.name == "Updated Name"
+    assert updated_author.dob == original_dob  # Should be unchanged
+    assert updated_author.id == original_id
+
+
+async def test_repo_update_many_with_model_instances_partial_fields_github_560(
+    seeded_test_session_async: "tuple[AsyncSession, dict[str, type]]",
+) -> None:
+    """Test repository update_many with model instances for partial updates (GitHub Issue #560).
+
+    This test verifies that update_many correctly handles model instances with partially
+    set fields, preventing None values from overwriting existing data.
+    """
+    _session, models = seeded_test_session_async
+    author_model = models["author"]
+    author_repo = get_repository_from_session(seeded_test_session_async, "author")
+
+    # Create multiple authors with all fields populated
+    author1 = await maybe_async(author_repo.create({"name": "Author One", "dob": datetime.date(1990, 1, 1)}))
+    author2 = await maybe_async(author_repo.create({"name": "Author Two", "dob": datetime.date(1991, 2, 2)}))
+
+    original_dob1 = author1.dob
+    original_dob2 = author2.dob
+
+    # Create partial updates using model instances with only id and name set
+    partial_updates = [
+        author_model(id=author1.id, name="Updated One"),
+        author_model(id=author2.id, name="Updated Two"),
+    ]
+
+    # Update via repository - should only update names, leave dobs unchanged
+    updated_authors = await maybe_async(author_repo.update_many(partial_updates))
+
+    # Verify: names were updated, but dobs remain unchanged
+    assert len(updated_authors) == 2
+    updated_by_id = {author.id: author for author in updated_authors}
+
+    assert updated_by_id[author1.id].name == "Updated One"
+    assert updated_by_id[author1.id].dob == original_dob1  # Should be unchanged
+
+    assert updated_by_id[author2.id].name == "Updated Two"
+    assert updated_by_id[author2.id].dob == original_dob2  # Should be unchanged

tests/unit/test_repository.py

@@ -1832,3 +1832,77 @@ def test_repository_update_methods_with_numpy_arrays() -> None:
     complex3 = np.array([1 + 2j, 5 + 6j])
     assert compare_values(complex1, complex2) is True
     assert compare_values(complex1, complex3) is False
+
+
+def test_was_attribute_set_with_explicitly_set_attributes() -> None:
+    """Test was_attribute_set correctly identifies explicitly set attributes."""
+    from sqlalchemy import inspect
+
+    from advanced_alchemy.repository._util import was_attribute_set
+
+    # Create an instance with explicitly set attributes
+    instance = UUIDModel()
+    instance.id = uuid4()  # Explicitly set id
+
+    # Get the mapper/inspector
+    mapper = inspect(instance)
+
+    # Explicitly set attributes should return True
+    assert was_attribute_set(instance, mapper, "id") is True
+
+
+def test_was_attribute_set_with_uninitialized_attributes() -> None:
+    """Test was_attribute_set correctly identifies uninitialized attributes."""
+    from sqlalchemy import inspect
+
+    from advanced_alchemy.repository._util import was_attribute_set
+
+    # Use the existing UUIDModel which has created_at and updated_at audit fields
+    # Create an instance - created_at and updated_at won't be in instance dict yet
+    instance = UUIDModel()
+
+    # Get the mapper/inspector
+    mapper = inspect(instance)
+
+    # Uninitialized audit attributes should return False
+    # They exist on the model but haven't been explicitly set
+    assert was_attribute_set(instance, mapper, "created_at") is False
+    assert was_attribute_set(instance, mapper, "updated_at") is False
+
+
+def test_was_attribute_set_with_modified_attributes() -> None:
+    """Test was_attribute_set detects attributes with modification history."""
+    from sqlalchemy import inspect
+
+    from advanced_alchemy.repository._util import was_attribute_set
+
+    # Create an instance and explicitly set attributes
+    instance = UUIDModel()
+    instance.id = uuid4()  # Explicitly set id
+
+    # Also test setting a datetime attribute
+    now = datetime.datetime.now(datetime.timezone.utc)
+    instance.created_at = now  # Explicitly modify created_at
+
+    # Get the mapper/inspector
+    mapper = inspect(instance)
+
+    # Modified attributes should return True
+    assert was_attribute_set(instance, mapper, "id") is True
+    assert was_attribute_set(instance, mapper, "created_at") is True
+
+
+def test_was_attribute_set_with_nonexistent_attribute() -> None:
+    """Test was_attribute_set handles nonexistent attributes gracefully."""
+    from sqlalchemy import inspect
+
+    from advanced_alchemy.repository._util import was_attribute_set
+
+    # Create an instance
+    instance = UUIDModel()
+
+    # Get the mapper/inspector
+    mapper = inspect(instance)
+
+    # Nonexistent attribute should return False (attr_state is None)
+    assert was_attribute_set(instance, mapper, "nonexistent_field") is False