Add replace_ignore ColumnInfo

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.4.0-alpha-13
+current_version = 0.4.0-alpha-14
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(-(?P<release>.*)-(?P<build>\d+))?
 serialize = 
 	{major}.{minor}.{patch}-{release}-{build}

pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "typing-extensions",
 ]
 name = "sql-athame"
-version = "0.4.0-alpha-13"
+version = "0.4.0-alpha-14"
 description = "Python tool for slicing and dicing SQL"
 readme = "README.md"
 

sql_athame/dataclasses.py

@@ -46,6 +46,7 @@ class ColumnInfo:
         serialize: Function to transform Python values before database storage
         deserialize: Function to transform database values back to Python objects
         insert_only: Whether this field should only be set on INSERT, not UPDATE in upsert operations
+        replace_ignore: Whether this field should be ignored for `replace_multiple`
 
     Example:
         >>> from dataclasses import dataclass
@@ -72,6 +73,7 @@ class ColumnInfo:
     serialize: Optional[Callable[[Any], Any]] = None
     deserialize: Optional[Callable[[Any], Any]] = None
     insert_only: Optional[bool] = None
+    replace_ignore: Optional[bool] = None
 
     def __post_init__(self, constraints: Union[str, Iterable[str], None]) -> None:
         if constraints is not None:
@@ -98,6 +100,9 @@ def merge(a: "ColumnInfo", b: "ColumnInfo") -> "ColumnInfo":
             serialize=b.serialize if b.serialize is not None else a.serialize,
             deserialize=b.deserialize if b.deserialize is not None else a.deserialize,
             insert_only=b.insert_only if b.insert_only is not None else a.insert_only,
+            replace_ignore=(
+                b.replace_ignore if b.replace_ignore is not None else a.replace_ignore
+            ),
         )
 
 
@@ -118,6 +123,7 @@ class ConcreteColumnInfo:
         serialize: Optional serialization function
         deserialize: Optional deserialization function
         insert_only: Whether this field should only be set on INSERT, not UPDATE
+        replace_ignore: Whether this field should be ignored for `replace_multiple`
     """
 
     field: Field
@@ -126,9 +132,10 @@ class ConcreteColumnInfo:
     create_type: str
     nullable: bool
     constraints: tuple[str, ...]
-    serialize: Optional[Callable[[Any], Any]] = None
-    deserialize: Optional[Callable[[Any], Any]] = None
-    insert_only: bool = False
+    serialize: Optional[Callable[[Any], Any]]
+    deserialize: Optional[Callable[[Any], Any]]
+    insert_only: bool
+    replace_ignore: bool
 
     @staticmethod
     def from_column_info(
@@ -163,6 +170,7 @@ def from_column_info(
             serialize=info.serialize,
             deserialize=info.deserialize,
             insert_only=bool(info.insert_only),
+            replace_ignore=bool(info.replace_ignore),
         )
 
     def create_table_string(self) -> str:
@@ -386,6 +394,20 @@ def insert_only_field_names(cls) -> set[str]:
             },
         )
 
+    @classmethod
+    def replace_ignore_field_names(cls) -> set[str]:
+        """Get set of field names marked as replace_ignore in ColumnInfo.
+
+        Returns:
+            Set of field names that should be ignored for `replace_multiple`
+        """
+        return cls._cached(
+            ("replace_ignore_field_names",),
+            lambda: {
+                ci.field.name for ci in cls.column_info().values() if ci.replace_ignore
+            },
+        )
+
     @classmethod
     def field_names_sql(
         cls,
@@ -1366,7 +1388,8 @@ async def plan_replace_multiple(
         """
         # For comparison purposes, combine auto-detected insert_only fields with manual ones
         all_insert_only = cls.insert_only_field_names() | set(insert_only)
-        ignore = sorted(set(ignore) | all_insert_only)
+        default_ignore = cls.replace_ignore_field_names() - set(force_update)
+        ignore = sorted(set(ignore) | default_ignore | all_insert_only)
         equal_ignoring = cls._cached(
             ("equal_ignoring", tuple(ignore)),
             lambda: cls._get_equal_ignoring_fn(ignore),
@@ -1512,7 +1535,8 @@ async def replace_multiple_reporting_differences(
         """
         # For comparison purposes, combine auto-detected insert_only fields with manual ones
         all_insert_only = cls.insert_only_field_names() | set(insert_only)
-        ignore = sorted(set(ignore) | all_insert_only)
+        default_ignore = cls.replace_ignore_field_names() - set(force_update)
+        ignore = sorted(set(ignore) | default_ignore | all_insert_only)
         differences_ignoring = cls._cached(
             ("differences_ignoring", tuple(ignore)),
             lambda: cls._get_differences_ignoring_fn(ignore),

tests/test_asyncpg.py

@@ -396,3 +396,171 @@ class Test(ModelBase, table_name="test_upsert", primary_key="id"):
     assert result[0].created_at == "2023-01-04"  # Should be updated
     assert result[0].name == "Alice Force"
     assert result[0].count == 20
+
+
+async def test_replace_multiple_with_replace_ignore(conn):
+    """Test replace_ignore ColumnInfo attribute."""
+
+    @dataclass(order=True)
+    class Test(ModelBase, table_name="test", primary_key="id"):
+        id: int
+        name: str
+        count: int
+        # metadata field should be ignored during comparison
+        metadata: Annotated[str, ColumnInfo(replace_ignore=True)]
+
+    await conn.execute(*Test.create_table_sql())
+
+    # Insert initial data
+    data = [
+        Test(1, "Alice", 10, "meta1"),
+        Test(2, "Bob", 20, "meta2"),
+        Test(3, "Charlie", 30, "meta3"),
+    ]
+    await Test.insert_multiple(conn, data)
+
+    # Replace with same data but different metadata
+    # Since metadata is ignored, no updates should happen
+    new_data = [
+        Test(1, "Alice", 10, "different_meta"),
+        Test(2, "Bob", 20, "different_meta"),
+        Test(3, "Charlie", 30, "different_meta"),
+    ]
+    c, u, d = await Test.replace_multiple(conn, new_data, where=[])
+    assert not c  # No creates
+    assert not u  # No updates because metadata is ignored
+    assert not d  # No deletes
+
+    # Verify original metadata is preserved
+    result = await Test.select(conn, order_by="id")
+    assert result[0].metadata == "meta1"
+    assert result[1].metadata == "meta2"
+    assert result[2].metadata == "meta3"
+
+    # Now change a non-ignored field - should trigger update
+    # The metadata will be updated too (it's only ignored for comparison)
+    new_data[0] = Test(1, "Alice Updated", 10, "still_different")
+    c, u, d = await Test.replace_multiple(conn, new_data, where=[])
+    assert not c
+    assert len(u) == 1  # Should update because name changed
+    assert not d
+
+    # Verify update happened - metadata gets updated along with other fields
+    result = await Test.select(conn, where=sql("id = 1"))
+    assert result[0].name == "Alice Updated"
+    assert result[0].metadata == "still_different"  # Updated along with name
+
+
+async def test_replace_multiple_replace_ignore_with_force_update(conn):
+    """Test that force_update overrides replace_ignore."""
+
+    @dataclass(order=True)
+    class Test(ModelBase, table_name="test", primary_key="id"):
+        id: int
+        name: str
+        metadata: Annotated[str, ColumnInfo(replace_ignore=True)]
+
+    await conn.execute(*Test.create_table_sql())
+
+    # Insert initial data
+    data = [Test(1, "Alice", "meta1"), Test(2, "Bob", "meta2")]
+    await Test.insert_multiple(conn, data)
+
+    # Replace with different metadata, using force_update
+    new_data = [Test(1, "Alice", "new_meta1"), Test(2, "Bob", "new_meta2")]
+    c, u, d = await Test.replace_multiple(
+        conn, new_data, where=[], force_update={"metadata"}
+    )
+    assert not c
+    assert len(u) == 2  # Should update because force_update overrides replace_ignore
+    assert not d
+
+    # Verify metadata was updated
+    result = await Test.select(conn, order_by="id")
+    assert result[0].metadata == "new_meta1"
+    assert result[1].metadata == "new_meta2"
+
+
+async def test_replace_multiple_replace_ignore_with_insert_only(conn):
+    """Test interaction between replace_ignore and insert_only."""
+
+    @dataclass(order=True)
+    class Test(ModelBase, table_name="test", primary_key="id"):
+        id: int
+        name: str
+        # Both replace_ignore and insert_only
+        created_at: Annotated[str, ColumnInfo(replace_ignore=True, insert_only=True)]
+        # Only replace_ignore
+        metadata: Annotated[str, ColumnInfo(replace_ignore=True)]
+
+    await conn.execute(*Test.create_table_sql())
+
+    # Insert initial data
+    data = [Test(1, "Alice", "2023-01-01", "meta1")]
+    await Test.insert_multiple(conn, data)
+
+    # Try to replace with different created_at and metadata
+    new_data = [Test(1, "Alice", "2023-01-02", "meta2")]
+    c, u, d = await Test.replace_multiple(conn, new_data, where=[])
+    assert not c
+    assert not u  # No update because both fields are ignored
+    assert not d
+
+    # Verify original values preserved
+    result = await Test.select(conn)
+    assert result[0].created_at == "2023-01-01"
+    assert result[0].metadata == "meta1"
+
+    # Change name - should trigger update
+    # created_at is preserved (insert_only), metadata is updated (only ignored for comparison)
+    new_data = [Test(1, "Alice Updated", "2023-01-03", "meta3")]
+    c, u, d = await Test.replace_multiple(conn, new_data, where=[])
+    assert not c
+    assert len(u) == 1
+    assert not d
+
+    # Verify update happened
+    result = await Test.select(conn)
+    assert result[0].name == "Alice Updated"
+    assert result[0].created_at == "2023-01-01"  # Preserved (insert_only)
+    assert result[0].metadata == "meta3"  # Updated (only ignored for comparison)
+
+
+async def test_replace_multiple_replace_ignore_partial_match(conn):
+    """Test replace_ignore when only some records match."""
+
+    @dataclass(order=True)
+    class Test(ModelBase, table_name="test", primary_key="id"):
+        id: int
+        category: str
+        value: int
+        metadata: Annotated[str, ColumnInfo(replace_ignore=True)]
+
+    await conn.execute(*Test.create_table_sql())
+
+    # Insert data with different categories
+    data = [
+        Test(1, "A", 10, "meta1"),
+        Test(2, "A", 20, "meta2"),
+        Test(3, "B", 30, "meta3"),
+    ]
+    await Test.insert_multiple(conn, data)
+
+    # Replace only category A with different metadata
+    new_data = [
+        Test(1, "A", 10, "new_meta1"),
+        Test(2, "A", 25, "new_meta2"),  # value changed
+    ]
+    c, u, d = await Test.replace_multiple(conn, new_data, where=sql("category = 'A'"))
+    assert not c
+    assert len(u) == 1  # Only id=2 should update (value changed)
+    assert not d  # Category B record not affected by where clause
+
+    # Verify results
+    result = await Test.select(conn, order_by="id")
+    assert len(result) == 3
+    assert result[0].metadata == "meta1"  # Unchanged (no update happened)
+    assert result[0].value == 10
+    assert result[1].metadata == "new_meta2"  # Updated along with value
+    assert result[1].value == 25  # Updated
+    assert result[2] == data[2]  # Category B unchanged