feat: match against complex types (#501)

Corrected an issue that prevented certain data types from working with the matching function.

Author: cofin

.gitignore

@@ -176,3 +176,4 @@ cython_debug/
 CLAUDE.md
 .gitignore
 .todos
+.tmp

advanced_alchemy/repository/_async.py

@@ -1,3 +1,5 @@
+import datetime
+import decimal
 import random
 import string
 from collections.abc import Iterable, Sequence
@@ -55,6 +57,18 @@
 
 DEFAULT_INSERTMANYVALUES_MAX_PARAMETERS: Final = 950
 POSTGRES_VERSION_SUPPORTING_MERGE: Final = 15
+DEFAULT_SAFE_TYPES: Final[set[type[Any]]] = {
+    int,
+    float,
+    str,
+    bool,
+    bytes,
+    decimal.Decimal,
+    datetime.date,
+    datetime.datetime,
+    datetime.time,
+    datetime.timedelta,
+}
 
 
 @runtime_checkable
@@ -497,6 +511,65 @@ def _get_uniquify(self, uniquify: Optional[bool] = None) -> bool:
         """
         return bool(uniquify) if uniquify is not None else self._uniquify
 
+    def _type_must_use_in_instead_of_any(self, matched_values: "list[Any]", field_type: "Any" = None) -> bool:
+        """Determine if field.in_() should be used instead of any_() for compatibility.
+
+        Uses SQLAlchemy's type introspection to detect types that may have DBAPI
+        serialization issues with the ANY() operator. Checks if actual values match
+        the column's expected python_type - mismatches indicate complex types that
+        need the safer IN() operator. Falls back to Python type checking when
+        SQLAlchemy type information is unavailable.
+
+        Args:
+            matched_values: Values to be used in the filter
+            field_type: Optional SQLAlchemy TypeEngine from the column
+
+        Returns:
+            bool: True if field.in_() should be used instead of any_()
+        """
+        if not matched_values:
+            return False
+
+        if field_type is not None:
+            try:
+                expected_python_type = getattr(field_type, "python_type", None)
+                if expected_python_type is not None:
+                    for value in matched_values:
+                        if value is not None and not isinstance(value, expected_python_type):
+                            return True
+            except (AttributeError, NotImplementedError):
+                return True
+
+        return any(value is not None and type(value) not in DEFAULT_SAFE_TYPES for value in matched_values)
+
+    def _get_unique_values(self, values: "list[Any]") -> "list[Any]":
+        """Get unique values from a list, handling unhashable types safely.
+
+        Args:
+            values: List of values to deduplicate
+
+        Returns:
+            list[Any]: List of unique values preserving order
+        """
+        if not values:
+            return []
+
+        try:
+            # Fast path for hashable types
+            seen: set[Any] = set()
+            unique_values: list[Any] = []
+            for value in values:
+                if value not in seen:
+                    unique_values.append(value)
+                    seen.add(value)
+        except TypeError:
+            # Fallback for unhashable types (e.g., dicts from JSONB)
+            unique_values = []
+            for value in values:
+                if value not in unique_values:
+                    unique_values.append(value)
+        return unique_values
+
     @staticmethod
     def _get_error_messages(
         error_messages: Optional[Union[ErrorMessages, EmptyType]] = Empty,
@@ -974,9 +1047,11 @@ def _get_delete_many_statement(
             statement = statement.execution_options(**execution_options)
         if supports_returning and statement_type != "select":
             statement = cast("ReturningDelete[tuple[ModelT]]", statement.returning(model_type))  # type: ignore[union-attr,assignment]  # pyright: ignore[reportUnknownLambdaType,reportUnknownMemberType,reportAttributeAccessIssue,reportUnknownVariableType]
-        if self._prefer_any:
-            return statement.where(any_(id_chunk) == id_attribute)  # type: ignore[arg-type]
-        return statement.where(id_attribute.in_(id_chunk))  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
+        # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+        use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(id_chunk, id_attribute.type)
+        if use_in:
+            return statement.where(id_attribute.in_(id_chunk))  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
+        return statement.where(any_(id_chunk) == id_attribute)  # type: ignore[arg-type]
 
     async def get(
         self,
@@ -1870,7 +1945,9 @@ async def upsert_many(
                 matched_values = [
                     field_data for datum in data if (field_data := getattr(datum, field_name)) is not None
                 ]
-                match_filter.append(any_(matched_values) == field if self._prefer_any else field.in_(matched_values))  # type: ignore[arg-type]
+                # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+                use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(matched_values, field.type)
+                match_filter.append(field.in_(matched_values) if use_in else any_(matched_values) == field)  # type: ignore[arg-type]
 
         with wrap_sqlalchemy_exception(
             error_messages=error_messages, dialect_name=self._dialect.name, wrap_exceptions=self.wrap_exceptions
@@ -1883,10 +1960,12 @@ async def upsert_many(
             )
             for field_name in match_fields:
                 field = get_instrumented_attr(self.model_type, field_name)
-                matched_values = list(
-                    {getattr(datum, field_name) for datum in existing_objs if datum},  # ensure the list is unique
-                )
-                match_filter.append(any_(matched_values) == field if self._prefer_any else field.in_(matched_values))  # type: ignore[arg-type]
+                # Safe deduplication that handles unhashable types (e.g., JSONB dicts)
+                all_values = [getattr(datum, field_name) for datum in existing_objs if datum]
+                matched_values = self._get_unique_values(all_values)
+                # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+                use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(matched_values, field.type)
+                match_filter.append(field.in_(matched_values) if use_in else any_(matched_values) == field)  # type: ignore[arg-type]
             existing_ids = self._get_object_ids(existing_objs=existing_objs)
             data = self._merge_on_match_fields(data, existing_objs, match_fields)
             for datum in data:

advanced_alchemy/repository/_sync.py

@@ -1,5 +1,7 @@
 # Do not edit this file directly. It has been autogenerated from
 # advanced_alchemy/repository/_async.py
+import datetime
+import decimal
 import random
 import string
 from collections.abc import Iterable, Sequence
@@ -56,6 +58,18 @@
 
 DEFAULT_INSERTMANYVALUES_MAX_PARAMETERS: Final = 950
 POSTGRES_VERSION_SUPPORTING_MERGE: Final = 15
+DEFAULT_SAFE_TYPES: Final[set[type[Any]]] = {
+    int,
+    float,
+    str,
+    bool,
+    bytes,
+    decimal.Decimal,
+    datetime.date,
+    datetime.datetime,
+    datetime.time,
+    datetime.timedelta,
+}
 
 
 @runtime_checkable
@@ -498,6 +512,65 @@ def _get_uniquify(self, uniquify: Optional[bool] = None) -> bool:
         """
         return bool(uniquify) if uniquify is not None else self._uniquify
 
+    def _type_must_use_in_instead_of_any(self, matched_values: "list[Any]", field_type: "Any" = None) -> bool:
+        """Determine if field.in_() should be used instead of any_() for compatibility.
+
+        Uses SQLAlchemy's type introspection to detect types that may have DBAPI
+        serialization issues with the ANY() operator. Checks if actual values match
+        the column's expected python_type - mismatches indicate complex types that
+        need the safer IN() operator. Falls back to Python type checking when
+        SQLAlchemy type information is unavailable.
+
+        Args:
+            matched_values: Values to be used in the filter
+            field_type: Optional SQLAlchemy TypeEngine from the column
+
+        Returns:
+            bool: True if field.in_() should be used instead of any_()
+        """
+        if not matched_values:
+            return False
+
+        if field_type is not None:
+            try:
+                expected_python_type = getattr(field_type, "python_type", None)
+                if expected_python_type is not None:
+                    for value in matched_values:
+                        if value is not None and not isinstance(value, expected_python_type):
+                            return True
+            except (AttributeError, NotImplementedError):
+                return True
+
+        return any(value is not None and type(value) not in DEFAULT_SAFE_TYPES for value in matched_values)
+
+    def _get_unique_values(self, values: "list[Any]") -> "list[Any]":
+        """Get unique values from a list, handling unhashable types safely.
+
+        Args:
+            values: List of values to deduplicate
+
+        Returns:
+            list[Any]: List of unique values preserving order
+        """
+        if not values:
+            return []
+
+        try:
+            # Fast path for hashable types
+            seen: set[Any] = set()
+            unique_values: list[Any] = []
+            for value in values:
+                if value not in seen:
+                    unique_values.append(value)
+                    seen.add(value)
+        except TypeError:
+            # Fallback for unhashable types (e.g., dicts from JSONB)
+            unique_values = []
+            for value in values:
+                if value not in unique_values:
+                    unique_values.append(value)
+        return unique_values
+
     @staticmethod
     def _get_error_messages(
         error_messages: Optional[Union[ErrorMessages, EmptyType]] = Empty,
@@ -975,9 +1048,11 @@ def _get_delete_many_statement(
             statement = statement.execution_options(**execution_options)
         if supports_returning and statement_type != "select":
             statement = cast("ReturningDelete[tuple[ModelT]]", statement.returning(model_type))  # type: ignore[union-attr,assignment]  # pyright: ignore[reportUnknownLambdaType,reportUnknownMemberType,reportAttributeAccessIssue,reportUnknownVariableType]
-        if self._prefer_any:
-            return statement.where(any_(id_chunk) == id_attribute)  # type: ignore[arg-type]
-        return statement.where(id_attribute.in_(id_chunk))  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
+        # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+        use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(id_chunk, id_attribute.type)
+        if use_in:
+            return statement.where(id_attribute.in_(id_chunk))  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
+        return statement.where(any_(id_chunk) == id_attribute)  # type: ignore[arg-type]
 
     def get(
         self,
@@ -1869,7 +1944,9 @@ def upsert_many(
                 matched_values = [
                     field_data for datum in data if (field_data := getattr(datum, field_name)) is not None
                 ]
-                match_filter.append(any_(matched_values) == field if self._prefer_any else field.in_(matched_values))  # type: ignore[arg-type]
+                # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+                use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(matched_values, field.type)
+                match_filter.append(field.in_(matched_values) if use_in else any_(matched_values) == field)  # type: ignore[arg-type]
 
         with wrap_sqlalchemy_exception(
             error_messages=error_messages, dialect_name=self._dialect.name, wrap_exceptions=self.wrap_exceptions
@@ -1882,10 +1959,12 @@ def upsert_many(
             )
             for field_name in match_fields:
                 field = get_instrumented_attr(self.model_type, field_name)
-                matched_values = list(
-                    {getattr(datum, field_name) for datum in existing_objs if datum},  # ensure the list is unique
-                )
-                match_filter.append(any_(matched_values) == field if self._prefer_any else field.in_(matched_values))  # type: ignore[arg-type]
+                # Safe deduplication that handles unhashable types (e.g., JSONB dicts)
+                all_values = [getattr(datum, field_name) for datum in existing_objs if datum]
+                matched_values = self._get_unique_values(all_values)
+                # Use field.in_() if types are incompatible with ANY() or if dialect doesn't prefer ANY()
+                use_in = not self._prefer_any or self._type_must_use_in_instead_of_any(matched_values, field.type)
+                match_filter.append(field.in_(matched_values) if use_in else any_(matched_values) == field)  # type: ignore[arg-type]
             existing_ids = self._get_object_ids(existing_objs=existing_objs)
             data = self._merge_on_match_fields(data, existing_objs, match_fields)
             for datum in data: