Support for cursor based pagination

doc/changelog.rst

@@ -6,12 +6,18 @@ Changelog
 
 Added
 ^^^^^
+- Support for :rfc:`RFC9865 <9865>`
 - :meth:`Resource.replace <scim2_models.Resource.replace>` returns whether the replacement
   changes the resource, the order of multi-valued entries aside. A server can keep
   ``meta.version`` and ``meta.lastModified`` when a PUT changes nothing.
 
 Changed
 ^^^^^^^
+- :meth:`SCIMException.from_error <scim2_models.SCIMException.from_error>` reconstructs
+  :class:`~scim2_models.InvalidCursorException`, :class:`~scim2_models.ExpiredCursorException` and
+  :class:`~scim2_models.InvalidCountException` from an :class:`~scim2_models.Error` carrying the
+  matching ``scimType``, as :rfc:`RFC9865 §2.1 <9865#section-2.1>` defines them. They used to fall
+  back to the base :class:`~scim2_models.SCIMException`.
 - :attr:`AttributeBinding.urn <scim2_models.AttributeBinding.urn>` and the error messages
   that quote it spell the attribute as the schema declares it, such as ``userName``, whatever
   case the path used.

scim2_models/__init__.py

@@ -22,6 +22,9 @@
 from .attributes import MultiValuedComplexAttribute
 from .base import BaseModel
 from .context import Context
+from .exceptions import ExpiredCursorException
+from .exceptions import InvalidCountException
+from .exceptions import InvalidCursorException
 from .exceptions import InvalidFilterException
 from .exceptions import InvalidPathException
 from .exceptions import InvalidSyntaxException
@@ -74,6 +77,7 @@
 from .resources.service_provider_config import ChangePassword
 from .resources.service_provider_config import ETag
 from .resources.service_provider_config import Filter
+from .resources.service_provider_config import Pagination
 from .resources.service_provider_config import Patch
 from .resources.service_provider_config import ServiceProviderConfig
 from .resources.service_provider_config import Sort
@@ -120,13 +124,16 @@
     "Entitlement",
     "Error",
     "ExtensibleStringEnum",
+    "ExpiredCursorException",
     "Extension",
     "External",
     "Filter",
     "Group",
     "GroupMember",
     "GroupMembership",
     "Im",
+    "InvalidCountException",
+    "InvalidCursorException",
     "InvalidFilterException",
     "InvalidPathException",
     "InvalidSyntaxException",
@@ -141,6 +148,7 @@
     "MultiValuedComplexAttribute",
     "Name",
     "NoTargetException",
+    "Pagination",
     "Patch",
     "PatchOp",
     "PatchOperation",

scim2_models/exceptions.py

@@ -314,6 +314,45 @@ class SensitiveException(SCIMException):
     )
 
 
+class InvalidCursorException(SCIMException):
+    """Cursor value is invalid.
+
+    Corresponds to scimType ``invalidCursor`` with HTTP status 400.
+
+    :rfc:`RFC 9865 Section 2.1  <9865#section-2.1>`
+    """
+
+    status = 400
+    scim_type = "invalidCursor"
+    _default_detail = "Cursor value is invalid. Cursor value SHOULD be empty to request the first page and set to the nextCursor or previousCursor value for subsequent queries."
+
+
+class ExpiredCursorException(SCIMException):
+    """Cursor has expired.
+
+    Corresponds to scimType ``expiredCursor`` with HTTP status 400.
+
+    :rfc:`RFC 9865 Section 2.3  <9865#section-2.3>`
+    """
+
+    status = 400
+    scim_type = "expiredCursor"
+    _default_detail = "Cursor has expired. Do not wait longer than service provider's cursorTimeout to request additional pages."
+
+
+class InvalidCountException(SCIMException):
+    """Count value is invalid.
+
+    Corresponds to scimType ``invalidCount`` with HTTP status 400.
+
+    :rfc:`RFC 9865 Section 2.4  <9865#section-2.4>`
+    """
+
+    status = 400
+    scim_type = "invalidCount"
+    _default_detail = "Count value is invalid. Count value must be between 0 and service provider's maxPageSize and must be equal to the count value of the initial query."
+
+
 _SCIM_TYPE_TO_EXCEPTION: dict[str, type[SCIMException]] = {
     "invalidFilter": InvalidFilterException,
     "tooMany": TooManyException,
@@ -325,4 +364,7 @@ class SensitiveException(SCIMException):
     "invalidValue": InvalidValueException,
     "invalidVers": InvalidVersionException,
     "sensitive": SensitiveException,
+    "invalidCursor": InvalidCursorException,
+    "expiredCursor": ExpiredCursorException,
+    "invalidCount": InvalidCountException,
 }

scim2_models/messages/list_response.py

@@ -1,14 +1,17 @@
+import re
 from typing import Any
 from typing import Generic
 from typing import Self
 
 from pydantic import Field
 from pydantic import ValidationInfo
 from pydantic import ValidatorFunctionWrapHandler
+from pydantic import field_validator
 from pydantic import model_validator
 from pydantic_core import PydanticCustomError
 
 from ..context import Context
+from ..exceptions import InvalidCursorException
 from ..resources.resource import AnyResource
 from ..urn import URN
 from .message import Message
@@ -53,6 +56,25 @@ class ListResponse(
     items_per_page: int | None = None
     """The number of resources returned in a list response page."""
 
+    next_cursor: str | None = None
+    """A string value that can be used to retrieve the next page of list
+    results."""
+
+    previous_cursor: str | None = None
+    """A string value that can be used to retrieve the previous page of list
+    results."""
+
+    @field_validator("next_cursor", "previous_cursor")
+    @classmethod
+    def validate_cursor_chars(cls, value: str | None) -> str | None:
+        """According to :rfc:`RFC9865 §2 <9865#section-2>`, cursor values may only contain unreserved characters as defined in :rfc:`RFC3986 §2.3 <3986#section-2.3>`.
+
+        An empty cursor requests the first page, so a response cannot carry one.
+        """
+        if value is not None and not re.fullmatch(r"[A-Za-z0-9\-._~]+", value):
+            raise InvalidCursorException().as_pydantic_error()
+        return value
+
     resources: list[AnyResource] | None = Field(None, serialization_alias="Resources")
     """A multi-valued list of complex objects containing the requested
     resources."""
@@ -68,6 +90,10 @@ def _check_results_number(
 
         - 'totalResults' is required
         - 'resources' must be set if 'totalResults' is non-zero.
+
+        RFC9865 §2 makes 'totalResults' optional with cursor pagination.
+        A response uses cursor pagination if the service provider supports it,
+        or if the response carries a cursor.
         """
         obj = handler(value)
         assert isinstance(obj, cls)
@@ -79,13 +105,23 @@ def _check_results_number(
         ):
             return obj
 
-        if obj.total_results is None:
+        config = info.context.get("scim_spc")
+        cursor_pagination = bool(
+            (config and config.pagination and config.pagination.cursor)
+            or obj.next_cursor is not None
+            or obj.previous_cursor is not None
+        )
+        if not cursor_pagination and obj.total_results is None:
             raise PydanticCustomError(
                 "required_error",
                 "Field 'totalResults' is required but value is missing or null",
             )
 
-        if obj.total_results > 0 and obj.resources is None:
+        if (
+            obj.total_results is not None
+            and obj.total_results > 0
+            and obj.resources is None
+        ):
             raise PydanticCustomError(
                 "no_resource_error",
                 "Field 'Resources' is missing or null but 'totalResults' is non-zero.",

scim2_models/messages/search_request.py

@@ -1,3 +1,4 @@
+import re
 from collections.abc import Iterable
 from enum import StrEnum
 from inspect import isclass
@@ -9,6 +10,7 @@
 from ..annotations import CaseExact
 from ..annotations import Mutability
 from ..base import BaseModel
+from ..exceptions import InvalidCursorException
 from ..exceptions import InvalidFilterException
 from ..exceptions import InvalidPathException
 from ..path import Path
@@ -222,6 +224,21 @@ def _start_index_floor(cls, value: int | None) -> int | None:
         """
         return None if value is None else max(1, value)
 
+    cursor: str | None = None
+    """A string value that can be used to retrieve the next page of results.
+    The cursor value is defined in :rfc:`RFC9865 §2 <9865#section-2>`."""
+
+    @field_validator("cursor")
+    @classmethod
+    def validate_cursor_chars(cls, value: str | None) -> str | None:
+        """According to :rfc:`RFC9865 §2 <9865#section-2>`, cursor values may only contain unreserved characters as defined in :rfc:`RFC3986 §2.3 <3986#section-2.3>`.
+
+        unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
+        """
+        if value is not None and not re.fullmatch(r"[A-Za-z0-9\-._~]*", value):
+            raise InvalidCursorException().as_pydantic_error()
+        return value
+
     count: int | None = None
     """An integer indicating the desired maximum number of query results per
     page."""

scim2_models/resources/service_provider_config.py

@@ -1,12 +1,15 @@
 from typing import Annotated
 from typing import Any
 
+from pydantic import field_validator
+
 from ..annotations import Mutability
 from ..annotations import Required
 from ..annotations import Returned
 from ..annotations import Uniqueness
 from ..attributes import ComplexAttribute
 from ..attributes import ExtensibleStringEnum
+from ..exceptions import InvalidValueException
 from ..reference import External
 from ..reference import Reference
 from ..urn import URN
@@ -52,6 +55,41 @@ class ETag(ComplexAttribute):
     """A Boolean value specifying whether or not the operation is supported."""
 
 
+class Pagination(ComplexAttribute):
+    class DefaultPaginationMethod(ExtensibleStringEnum):
+        cursor = "cursor"
+        index = "index"  # type: ignore[assignment]
+
+    cursor: Annotated[bool | None, Mutability.read_only, Required.true] = None
+    """A Boolean value specifying support of cursor-based pagination."""
+
+    index: Annotated[bool | None, Mutability.read_only, Required.true] = None
+    """A Boolean value specifying support of index-based pagination."""
+
+    default_pagination_method: Annotated[
+        DefaultPaginationMethod | None, Mutability.read_only
+    ] = None
+    """A string value specifying the type of pagination that the service provider defaults to when the client has not specified which method it wishes to use. Possible values are "cursor" and "index"."""
+
+    default_page_size: Annotated[int | None, Mutability.read_only] = None
+    """Positive integer value specifying the default number of results returned in a page when a count is not specified in the query."""
+
+    max_page_size: Annotated[int | None, Mutability.read_only] = None
+    """Positive integer specifying the maximum number of results returned in a page regardless of what is specified for the count in a query. The maximum number of results returned may be further restricted by other criteria."""
+
+    cursor_timeout: Annotated[int | None, Mutability.read_only] = None
+    """Positive integer specifying the minimum number of seconds that a cursor is valid between page requests. Clients waiting too long between cursor pagination requests may receive an invalid cursor error response. No value being specified may mean that there is no cursor timeout or that the cursor timeout is not a static duration."""
+
+    @field_validator("default_page_size", "max_page_size", "cursor_timeout")
+    @classmethod
+    def validate_positive_integers(cls, value: int | None) -> int | None:
+        if value is not None and value <= 0:
+            raise InvalidValueException(
+                detail=f"{value} is not a positive integer"
+            ).as_pydantic_error()
+        return value
+
+
 class AuthenticationScheme(ComplexAttribute):
     class Type(ExtensibleStringEnum):
         oauth = "oauth"
@@ -130,3 +168,6 @@ class ServiceProviderConfig(Resource[Any]):
     ] = None
     """A complex type that specifies supported authentication scheme
     properties."""
+
+    pagination: Annotated[Pagination | None, Mutability.read_only] = None
+    """A complex type that specifies pagination configuration options."""

tests/test_dynamic_schemas.py

@@ -99,7 +99,9 @@ def test_dynamic_service_provider_config_schema(load_sample):
     canonic_schema(sample)
 
     schema["attributes"] = [
-        attr for attr in schema["attributes"] if attr["name"] != "id"
+        attr
+        for attr in schema["attributes"]
+        if attr["name"] not in ("id", "pagination")
     ]
 
     assert sample == schema

tests/test_errors.py

@@ -1,3 +1,6 @@
+from scim2_models.exceptions import ExpiredCursorException
+from scim2_models.exceptions import InvalidCountException
+from scim2_models.exceptions import InvalidCursorException
 from scim2_models.exceptions import InvalidFilterException
 from scim2_models.exceptions import InvalidPathException
 from scim2_models.exceptions import InvalidSyntaxException
@@ -23,5 +26,8 @@ def test_predefined_errors():
         InvalidValueException(),
         InvalidVersionException(),
         SensitiveException(),
+        InvalidCursorException(),
+        ExpiredCursorException(),
+        InvalidCountException(),
     ):
         assert isinstance(exc.to_error(), Error)