fix: support SQLAlchemy 2.1

Author: azmeuk

src/scim2_sqlalchemy/mapping.py

@@ -49,12 +49,21 @@ class _Storage(Enum):
     link_type = "link type"
 
 
-def _holds_text(expression: Any) -> bool:
-    """Whether an expression holds strings. A type that tells no Python type is taken as text."""
+def _python_type(expression: Any) -> type | None:
+    """Return the Python type of the values of an expression, or None if its type tells none.
+
+    SQLAlchemy 2.0 raises NotImplementedError for such a type, and 2.1 returns object.
+    """
     try:
-        return expression.type.python_type is str
+        python_type: type = expression.type.python_type
     except NotImplementedError:
-        return True
+        return None
+    return None if python_type is object else python_type
+
+
+def _holds_text(expression: Any) -> bool:
+    """Whether an expression holds strings. A type that tells no Python type is taken as text."""
+    return _python_type(expression) in (str, None)
 
 
 class Attribute:

src/scim2_sqlalchemy/query.py

@@ -1,6 +1,7 @@
 import operator
 from collections.abc import Callable
 from typing import Any
+from typing import TypeVarTuple
 
 from scim2_models import AttributeBinding
 from scim2_models import InvalidCursorException
@@ -46,6 +47,7 @@
 from .mapping import ResourceMapping
 from .mapping import _Collection
 from .mapping import _Column
+from .mapping import _python_type
 
 _ORDERINGS: dict[CompareOperator, Callable[[Any, Any], Any]] = {
     CompareOperator.eq: operator.eq,
@@ -56,11 +58,13 @@
 }
 
 
+_Row = TypeVarTuple("_Row")
+
+
 def _as_key(column: _Column, value: Any) -> Any:
     """Return an identifier in the type of its column, or None when it cannot have that type."""
-    try:
-        python_type = column.expression.type.python_type
-    except NotImplementedError:
+    python_type = _python_type(column.expression)
+    if python_type is None:
         return value
     try:
         return python_type(value)
@@ -381,14 +385,18 @@ def _where(
     return _FilterTranslator(mapping, scim_filter, comparator).visit(scim_filter.ast)
 
 
-def _cursor_page(page: Select[Any], search_request: SearchRequest[Any]) -> Select[Any]:
+def _cursor_page(
+    page: Select[*_Row], search_request: SearchRequest[Any]
+) -> Select[*_Row]:
     """Limit a cursor page to count + 1 rows, the last one telling whether a page follows."""
     if search_request.count is None:
         return page
     return page.limit(search_request.count + 1)
 
 
-def _index_page(page: Select[Any], search_request: SearchRequest[Any]) -> Select[Any]:
+def _index_page(
+    page: Select[*_Row], search_request: SearchRequest[Any]
+) -> Select[*_Row]:
     """Limit a page to the records from startIndex, at most count of them."""
     page = page.offset(search_request.start_index_0 or 0)
     if search_request.count is None:
@@ -401,7 +409,7 @@ def _search_statements(
     search_request: SearchRequest[Any],
     comparator: _Comparator,
     anchor: _Anchor | None = None,
-) -> tuple[Select[tuple[int]], Select[tuple[Any, Any, str]]]:
+) -> tuple[Select[int], Select[Any, Any, str]]:
     """Return the statement counting the matching records, and the one selecting a page of them.
 
     The page lists each record with its sort value and its identifier as
@@ -443,7 +451,7 @@ def _root_statements(
     search_request: SearchRequest[Any],
     comparator: _Comparator,
     anchor: _Anchor | None = None,
-) -> tuple[Select[tuple[int]], Select[tuple[int, str, Any]]]:
+) -> tuple[Select[int], Select[int, str, Any]]:
     """Return the statement counting the matching records of several mappings, and the one selecting a page of them.
 
     The page lists the position of the mapping, the identifier of each
@@ -496,7 +504,7 @@ def _root_statements(
 
 def _records_statement(
     mapping: ResourceMapping, ids: list[str], parameters: ResponseParameters[Any]
-) -> Select[tuple[str, Any]]:
+) -> Select[str, Any]:
     """Return the statement loading the records of a page, each with its identifier as text."""
     keys = [_as_key(mapping._id, value) for value in ids]
     return (

tests/test_mapping.py

@@ -39,6 +39,17 @@ class Opaque(UserDefinedType[str]):
     cache_ok = True
 
 
+class Unknown(UserDefinedType[str]):
+    """A column type refusing to tell its Python type, as SQLAlchemy 2.0 types do."""
+
+    cache_ok = True
+
+    @property
+    def python_type(self) -> type:
+        """Refuse to tell the Python type."""
+        raise NotImplementedError
+
+
 class Person(Base):
     __tablename__ = "people"
 
@@ -50,6 +61,7 @@ class Person(Base):
     other_version: Mapped[int] = mapped_column()
     emails: Mapped[list["Address"]] = relationship()
     opaque: Mapped[str] = mapped_column(Opaque())
+    unknown: Mapped[str] = mapped_column(Unknown())
 
     __mapper_args__ = {"version_id_col": version}
 
@@ -182,12 +194,14 @@ def test_a_link_displays_nothing_without_a_name():
     assert groups._display() is None
 
 
-def test_a_column_type_without_python_type_holds_text():
+@pytest.mark.parametrize("record_column", ["opaque", "unknown"])
+def test_a_column_type_without_python_type_holds_text(record_column):
     """A column whose type tells no Python type is compared as it is."""
-    mapping = person_mapping(nickName=Person.opaque)
+    expression = getattr(Person, record_column)
+    mapping = person_mapping(nickName=expression)
     column = mapping._column("nickName")
 
-    assert _holds_text(Person.opaque)
+    assert _holds_text(expression)
     assert _as_key(column, "x") == "x"