Support for RFC 9865 cursors

doc/changelog.rst

@@ -14,8 +14,9 @@ Added
 - :class:`~scim2_sqlalchemy.SqlAlchemyStorage` and
   :class:`~scim2_sqlalchemy.AsyncSqlAlchemyStorage`, which serve the mappings to
   :doc:`scim2-server <scim2_server:index>`. They filter, sort and page the resources in the
-  database, on one resource type or at the root of the server. The asynchronous storage needs
-  the ``asyncio`` extra.
+  database, on one resource type or at the root of the server. They page with ``startIndex``, or
+  with the cursors of :rfc:`9865`, which give stable pages. The asynchronous storage needs the
+  ``asyncio`` extra.
 - :meth:`ResourceMapping.schemas <scim2_sqlalchemy.ResourceMapping.schemas>`, which publishes the
   attributes a mapping stores.
 - Filters, sorts and uniqueness checks compare strings with the comparison key of the policy of

doc/explanation.rst

@@ -103,6 +103,31 @@ the identifier of each resource of the page. It then loads the records of each t
 type without the sort attribute sorts its resources as having no value. Between two resources of
 different types that share a sort value, the order of the resource types decides.
 
+Cursors give stable pages
+-------------------------
+
+The storage pages with ``startIndex``, and with the cursors of :rfc:`9865`. With a cursor, the
+position of a page holds the sort value and the identifier of a resource: the last one of the
+page for the next page, and the first one for the previous page. The next page selects the
+resources after them in the sort order. A resource created or deleted between two pages does not
+move the other ones. :doc:`scim2_server:explanation/pagination` explains what these stable pages
+guarantee, and :doc:`scim2_server:how-to/page-with-cursors` explains how to announce the
+cursors.
+
+The sort value of a position is the value the database compares, after the comparison key of the
+policy. The storage writes the condition on the sort value by hand, since a row comparison would
+not place ``NULL`` where the sort does. At the root, the position also holds the resource type,
+which orders the resources of different types that share a sort value.
+
+The storage reads one resource more than ``count``, to know whether a page follows. It only knows
+it in the direction it reads: forward for a next page, backward for a previous page. In the other
+direction, it assumes that the page the client comes from still exists. When all the resources of
+that page were deleted meanwhile, the client gets an empty page without cursors, and starts again
+from the first page.
+
+Each page shows the resources as they are when it is read. The storage does not give a snapshot
+of the collection as it was on the first page.
+
 Writes that cannot be stored are refused
 ----------------------------------------
 
@@ -173,9 +198,6 @@ send the same queries.
 Limits
 ------
 
-The storage pages with ``startIndex`` and ``count``. It does not support the cursors of
-:rfc:`9865`.
-
 A search or a read only loads the collections its response keeps, following the ``attributes``
 and ``excludedAttributes`` parameters. With ``excludedAttributes=members``, the members of the
 groups are not loaded. The storage still loads every column of the records, which costs little

pyproject.toml

@@ -15,8 +15,8 @@ authors = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "scim2-models>=0.12.1",
-    "scim2-server>=0.7.1",
+    "scim2-models>=0.12.2",
+    "scim2-server>=0.8.0",
     "sqlalchemy>=2.0.41",
 ]
 
@@ -41,7 +41,7 @@ dev = [
     "pytest-cov>=7.1.0",
     "pytest-postgresql>=9.1.0",
     "scim2-client>=0.12.0",
-    "scim2-server[testing]>=0.7.1",
+    "scim2-server[testing]>=0.8.0",
     "scim2-tester>=0.5.2",
     "tox-uv>=1.36.0",
     "werkzeug>=3.1.8",

src/scim2_sqlalchemy/keyset.py

@@ -0,0 +1,143 @@
+import datetime
+import decimal
+import uuid
+from collections.abc import Callable
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from scim2_models import InvalidCursorException
+from sqlalchemy import ColumnElement
+from sqlalchemy import and_
+from sqlalchemy import false
+from sqlalchemy import nulls_first
+from sqlalchemy import nulls_last
+from sqlalchemy import or_
+from sqlalchemy import tuple_
+
+_TAGGED: dict[str, tuple[type, Callable[[Any], str], Callable[[str], Any]]] = {
+    "datetime": (
+        datetime.datetime,
+        datetime.datetime.isoformat,
+        datetime.datetime.fromisoformat,
+    ),
+    "date": (datetime.date, datetime.date.isoformat, datetime.date.fromisoformat),
+    "decimal": (decimal.Decimal, str, decimal.Decimal),
+    "uuid": (uuid.UUID, str, uuid.UUID),
+}
+
+
+def _encode(value: Any) -> Any:
+    """Return a sort value as a value JSON can hold, tagged with its type when JSON has none."""
+    if value is None or isinstance(value, bool | int | float | str):
+        return value
+    for tag, (python_type, dump, _) in _TAGGED.items():
+        if isinstance(value, python_type):
+            return {tag: dump(value)}
+    raise TypeError(f"Cannot page on a sort value of type {type(value).__name__}")
+
+
+def _decode(value: Any) -> Any:
+    """Return the sort value a position holds."""
+    if value is None or isinstance(value, bool | int | float | str):
+        return value
+    if not isinstance(value, dict) or len(value) != 1:
+        raise InvalidCursorException
+    ((tag, text),) = value.items()
+    if tag not in _TAGGED or not isinstance(text, str):
+        raise InvalidCursorException
+    try:
+        return _TAGGED[tag][2](text)
+    except (ValueError, decimal.InvalidOperation):
+        raise InvalidCursorException from None
+
+
+@dataclass(frozen=True)
+class _Anchor:
+    """The row a cursor page starts after, or ends before.
+
+    ties are the values that order the rows sharing a sort value: the
+    identifier as text, preceded at the root by the position of the resource
+    type.
+    """
+
+    forward: bool
+    key: Any
+    ties: tuple[Any, ...]
+
+
+def _anchor(position: Any, root: bool) -> _Anchor | None:
+    """Read the position a storage gave, or return None for the first page."""
+    if position is None:
+        return None
+    names = {"d", "k", "t", "i"} if root else {"d", "k", "i"}
+    if not isinstance(position, dict) or set(position) != names:
+        raise InvalidCursorException
+    if position["d"] not in ("next", "previous") or not isinstance(position["i"], str):
+        raise InvalidCursorException
+    ties: tuple[Any, ...] = (position["i"],)
+    if root:
+        if type(position["t"]) is not int:
+            raise InvalidCursorException
+        ties = (position["t"], position["i"])
+    return _Anchor(position["d"] == "next", _decode(position["k"]), ties)
+
+
+def _position(direction: str, key: Any, ties: Sequence[Any]) -> Any:
+    """Return the position of the page before or after a row, as a value JSON can hold.
+
+    The sort value is null when the search has no sort.
+    """
+    position: dict[str, Any] = {"d": direction, "k": _encode(key)}
+    if len(ties) == 2:
+        position["t"] = ties[0]
+    position["i"] = ties[-1]
+    return position
+
+
+def _key_order(key: Any, ascending: bool) -> Any:
+    """Per RFC 7644 §3.4.2.3, resources without a value come last when ascending and first when descending."""
+    if ascending:
+        return nulls_last(key.asc())
+    return nulls_first(key.desc())
+
+
+def _keyset_order(
+    key: Any, ties: Sequence[Any], descending: bool, forward: bool
+) -> list[Any]:
+    """Return the ORDER BY terms of a search: the sort value, then the ties.
+
+    A previous page is read backward, in the reverse order of the sort.
+    """
+    terms = [] if key is None else [_key_order(key, descending != forward)]
+    return [*terms, *(tie.asc() if forward else tie.desc() for tie in ties)]
+
+
+def _beyond(
+    key: Any,
+    ties: Sequence[Any],
+    anchor: _Anchor,
+    values: Sequence[Any],
+    descending: bool,
+) -> ColumnElement[bool]:
+    """Return the condition keeping the rows past the anchor, in the reading direction.
+
+    values are the ties of the anchor, in the types of the ties. The
+    condition on the sort value is written out, as a row comparison would not
+    place NULL where the sort does.
+    """
+    tie_after: ColumnElement[bool]
+    if len(ties) == 1:
+        tie_after = ties[0] > values[0] if anchor.forward else ties[0] < values[0]
+    else:
+        row, bound = tuple_(*ties), tuple_(*values)
+        tie_after = row > bound if anchor.forward else row < bound
+    if key is None:
+        return tie_after
+
+    ascending = descending != anchor.forward
+    if anchor.key is None:
+        key_after = key.is_not(None) if not ascending else false()
+        return or_(key_after, and_(key.is_(None), tie_after))
+    key_after = or_(key > anchor.key, key.is_(None)) if ascending else key < anchor.key
+    return or_(key_after, and_(key == anchor.key, tie_after))

src/scim2_sqlalchemy/query.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from scim2_models import AttributeBinding
+from scim2_models import InvalidCursorException
 from scim2_models import InvalidFilterException
 from scim2_models import InvalidPathException
 from scim2_models import ResponseParameters
@@ -29,7 +30,6 @@
 from sqlalchemy import literal
 from sqlalchemy import not_
 from sqlalchemy import null
-from sqlalchemy import nulls_first
 from sqlalchemy import nulls_last
 from sqlalchemy import or_
 from sqlalchemy import select
@@ -40,6 +40,9 @@
 from .comparison import _SUBSTRINGS
 from .comparison import _Comparator
 from .conversion import _as_stored
+from .keyset import _Anchor
+from .keyset import _beyond
+from .keyset import _keyset_order
 from .mapping import ResourceMapping
 from .mapping import _Collection
 from .mapping import _Column
@@ -270,28 +273,6 @@ def _sort_key(
     return expression
 
 
-def _sorted(expression: Any, search_request: SearchRequest[Any]) -> Any:
-    """Per RFC 7644 §3.4.2.3, resources without a value come last when ascending and first when descending."""
-    if search_request.sort_order == SearchRequest.SortOrder.descending:
-        return nulls_first(expression.desc())
-    return nulls_last(expression.asc())
-
-
-def _order_by(
-    mapping: ResourceMapping,
-    search_request: SearchRequest[Any],
-    comparator: _Comparator,
-) -> list[Any]:
-    """Return the ORDER BY terms of a search, the record identifier last.
-
-    Without the identifier, records sharing a sort value could change pages
-    from a request to the next.
-    """
-    key = _sort_key(mapping, search_request, comparator)
-    terms = [] if key is None else [_sorted(key, search_request)]
-    return [*terms, mapping._id.expression]
-
-
 def _first_entry(collection: _Collection, expression: Any) -> Any:
     """Return the value of the primary entry of a collection, or else of its first one."""
     relationship = collection.relationship_property
@@ -321,39 +302,74 @@ def _where(
     return _FilterTranslator(mapping, scim_filter, comparator).visit(scim_filter.ast)
 
 
+def _cursor_page(page: Select[Any], search_request: SearchRequest[Any]) -> Select[Any]:
+    """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]:
+    page = page.offset(search_request.start_index_0 or 0)
+    if search_request.count is None:
+        return page
+    return page.limit(search_request.count)
+
+
 def _search_statements(
     mapping: ResourceMapping,
     search_request: SearchRequest[Any],
     comparator: _Comparator,
-) -> tuple[Select[tuple[int]], Select[Any]]:
-    """Return the statement counting the matching records, and the one selecting a page of them."""
+    anchor: _Anchor | None = None,
+) -> tuple[Select[tuple[int]], Select[tuple[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
+    text. The records are ordered by sort value, then by identifier: without
+    the identifier, records sharing a sort value could change pages from a
+    request to the next. A cursor page is read from the anchor, in the reverse
+    order for a previous page.
+    """
     condition = _where(mapping, search_request, comparator)
     count = select(func.count()).select_from(mapping.record).where(condition)
+    key = _sort_key(mapping, search_request, comparator)
+    record_id = mapping._id.expression
+    descending = search_request.sort_order == SearchRequest.SortOrder.descending
+    forward = anchor is None or anchor.forward
     page = (
-        select(mapping.record)
+        select(
+            mapping.record,
+            (null() if key is None else key).label("key"),
+            cast(record_id, String).label("id"),
+        )
         .where(condition)
-        .order_by(*_order_by(mapping, search_request, comparator))
+        .order_by(*_keyset_order(key, [record_id], descending, forward))
         .options(
             *mapping._loader_options(mapping._returned_collections(search_request))
         )
-        .offset(search_request.start_index_0 or 0)
     )
-    if search_request.count is not None:
-        page = page.limit(search_request.count)
-    return count, page
+    if search_request.cursor is None:
+        return count, _index_page(page, search_request)
+    if anchor is not None:
+        value = _as_key(mapping._id, anchor.ties[0])
+        if value is None:
+            raise InvalidCursorException
+        page = page.where(_beyond(key, [record_id], anchor, [value], descending))
+    return count, _cursor_page(page, search_request)
 
 
 def _root_statements(
     mappings: list[ResourceMapping],
     search_request: SearchRequest[Any],
     comparator: _Comparator,
-) -> tuple[Select[tuple[int]], Select[tuple[int, str]]]:
+    anchor: _Anchor | None = None,
+) -> tuple[Select[tuple[int]], Select[tuple[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 and the identifier of each
-    record, as text, so that the identifiers of every table fit in one column.
-    A mapping whose model does not declare the sort attribute sorts its
-    records as having no value.
+    The page lists the position of the mapping, the identifier of each
+    record as text, so that the identifiers of every table fit in one column,
+    and its sort value. A mapping whose model does not declare the sort
+    attribute sorts its records as having no value.
     """
     conditions = [_where(mapping, search_request, comparator) for mapping in mappings]
     keys = [_sort_key(mapping, search_request, comparator) for mapping in mappings]
@@ -382,15 +398,20 @@ def _root_statements(
             columns.append(key.label("key"))
         branches.append(select(*columns).where(condition))
     rows = union_all(*branches).subquery()
-    terms = [] if key_type is None else [_sorted(rows.c.key, search_request)]
-    page = (
-        select(rows.c.position, rows.c.id)
-        .order_by(*terms, rows.c.position, rows.c.id)
-        .offset(search_request.start_index_0 or 0)
-    )
-    if search_request.count is not None:
-        page = page.limit(search_request.count)
-    return count, page
+    row_key = None if key_type is None else rows.c.key
+    ties = [rows.c.position, rows.c.id]
+    descending = search_request.sort_order == SearchRequest.SortOrder.descending
+    forward = anchor is None or anchor.forward
+    page = select(
+        rows.c.position, rows.c.id, (null() if row_key is None else row_key)
+    ).order_by(*_keyset_order(row_key, ties, descending, forward))
+    if search_request.cursor is None:
+        return count, _index_page(page, search_request)
+    if anchor is not None:
+        if anchor.ties[0] not in range(len(mappings)):
+            raise InvalidCursorException
+        page = page.where(_beyond(row_key, ties, anchor, anchor.ties, descending))
+    return count, _cursor_page(page, search_request)
 
 
 def _records_statement(