doc: suppose that database perform pagination themlselves

Author: azmeuk

doc/guides/_examples/django_example.py

@@ -23,7 +23,9 @@
 from .integrations import delete_record
 from .integrations import from_scim_user
 from .integrations import get_record
+from .integrations import get_resource_type
 from .integrations import get_resource_types
+from .integrations import get_schema
 from .integrations import get_schemas
 from .integrations import list_records
 from .integrations import save_record
@@ -170,11 +172,10 @@ def get(self, request):
         except ValidationError as error:
             return scim_validation_error(error)
 
-        all_records = list_records()
-        page = all_records[req.start_index_0 : req.stop_index_0]
+        total, page = list_records(req.start_index_0, req.stop_index_0)
         resources = [to_scim_user(record) for record in page]
         response = ListResponse[User](
-            total_results=len(all_records),
+            total_results=total,
             start_index=req.start_index or 1,
             items_per_page=len(resources),
             resources=resources,
@@ -227,10 +228,9 @@ def get(self, request):
         except ValidationError as error:
             return scim_validation_error(error)
 
-        all_schemas = get_schemas()
-        page = all_schemas[req.start_index_0 : req.stop_index_0]
+        total, page = get_schemas(req.start_index_0, req.stop_index_0)
         response = ListResponse[Schema](
-            total_results=len(all_schemas),
+            total_results=total,
             start_index=req.start_index or 1,
             items_per_page=len(page),
             resources=page,
@@ -244,13 +244,14 @@ class SchemaView(View):
     """Handle GET on a single SCIM schema."""
 
     def get(self, request, schema_id):
-        for schema in get_schemas():
-            if schema.id == schema_id:
-                return scim_response(
-                    schema.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
-                )
-        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
-        return scim_response(scim_error.model_dump_json(), HTTPStatus.NOT_FOUND)
+        try:
+            schema = get_schema(schema_id)
+        except KeyError:
+            scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
+            return scim_response(scim_error.model_dump_json(), HTTPStatus.NOT_FOUND)
+        return scim_response(
+            schema.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
+        )
 # -- schemas-end --
 
 
@@ -264,10 +265,9 @@ def get(self, request):
         except ValidationError as error:
             return scim_validation_error(error)
 
-        all_resource_types = get_resource_types()
-        page = all_resource_types[req.start_index_0 : req.stop_index_0]
+        total, page = get_resource_types(req.start_index_0, req.stop_index_0)
         response = ListResponse[ResourceType](
-            total_results=len(all_resource_types),
+            total_results=total,
             start_index=req.start_index or 1,
             items_per_page=len(page),
             resources=page,
@@ -281,15 +281,16 @@ class ResourceTypeView(View):
     """Handle GET on a single SCIM resource type."""
 
     def get(self, request, resource_type_id):
-        for rt in get_resource_types():
-            if rt.id == resource_type_id:
-                return scim_response(
-                    rt.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
-                )
-        scim_error = Error(
-            status=404, detail=f"ResourceType {resource_type_id!r} not found"
+        try:
+            rt = get_resource_type(resource_type_id)
+        except KeyError:
+            scim_error = Error(
+                status=404, detail=f"ResourceType {resource_type_id!r} not found"
+            )
+            return scim_response(scim_error.model_dump_json(), HTTPStatus.NOT_FOUND)
+        return scim_response(
+            rt.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
         )
-        return scim_response(scim_error.model_dump_json(), HTTPStatus.NOT_FOUND)
 # -- resource-types-end --
 
 

doc/guides/_examples/flask_example.py

@@ -20,7 +20,9 @@
 from .integrations import delete_record
 from .integrations import from_scim_user
 from .integrations import get_record
+from .integrations import get_resource_type
 from .integrations import get_resource_types
+from .integrations import get_schema
 from .integrations import get_schemas
 from .integrations import list_records
 from .integrations import save_record
@@ -165,11 +167,10 @@ def delete_user(app_record):
 def list_users():
     """Return one page of users as a SCIM ListResponse."""
     req = SearchRequest.model_validate(request.args.to_dict())
-    all_records = list_records()
-    page = all_records[req.start_index_0 : req.stop_index_0]
+    total, page = list_records(req.start_index_0, req.stop_index_0)
     resources = [to_scim_user(record) for record in page]
     response = ListResponse[User](
-        total_results=len(all_records),
+        total_results=total,
         start_index=req.start_index or 1,
         items_per_page=len(resources),
         resources=resources,
@@ -211,10 +212,9 @@ def create_user():
 def list_schemas():
     """Return one page of SCIM schemas the server exposes."""
     req = SearchRequest.model_validate(request.args.to_dict())
-    all_schemas = get_schemas()
-    page = all_schemas[req.start_index_0 : req.stop_index_0]
+    total, page = get_schemas(req.start_index_0, req.stop_index_0)
     response = ListResponse[Schema](
-        total_results=len(all_schemas),
+        total_results=total,
         start_index=req.start_index or 1,
         items_per_page=len(page),
         resources=page,
@@ -226,16 +226,17 @@ def list_schemas():
 
 
 @bp.get("/Schemas/<path:schema_id>")
-def get_schema(schema_id):
+def get_schema_by_id(schema_id):
     """Return one SCIM schema by its URI identifier."""
-    for schema in get_schemas():
-        if schema.id == schema_id:
-            return (
-                schema.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
-                HTTPStatus.OK,
-            )
-    scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
-    return scim_error.model_dump_json(), HTTPStatus.NOT_FOUND
+    try:
+        schema = get_schema(schema_id)
+    except KeyError:
+        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
+        return scim_error.model_dump_json(), HTTPStatus.NOT_FOUND
+    return (
+        schema.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
+        HTTPStatus.OK,
+    )
 # -- schemas-end --
 
 
@@ -244,10 +245,9 @@ def get_schema(schema_id):
 def list_resource_types():
     """Return one page of SCIM resource types the server exposes."""
     req = SearchRequest.model_validate(request.args.to_dict())
-    all_resource_types = get_resource_types()
-    page = all_resource_types[req.start_index_0 : req.stop_index_0]
+    total, page = get_resource_types(req.start_index_0, req.stop_index_0)
     response = ListResponse[ResourceType](
-        total_results=len(all_resource_types),
+        total_results=total,
         start_index=req.start_index or 1,
         items_per_page=len(page),
         resources=page,
@@ -259,18 +259,19 @@ def list_resource_types():
 
 
 @bp.get("/ResourceTypes/<resource_type_id>")
-def get_resource_type(resource_type_id):
+def get_resource_type_by_id(resource_type_id):
     """Return one SCIM resource type by its identifier."""
-    for rt in get_resource_types():
-        if rt.id == resource_type_id:
-            return (
-                rt.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
-                HTTPStatus.OK,
-            )
-    scim_error = Error(
-        status=404, detail=f"ResourceType {resource_type_id!r} not found"
+    try:
+        rt = get_resource_type(resource_type_id)
+    except KeyError:
+        scim_error = Error(
+            status=404, detail=f"ResourceType {resource_type_id!r} not found"
+        )
+        return scim_error.model_dump_json(), HTTPStatus.NOT_FOUND
+    return (
+        rt.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
+        HTTPStatus.OK,
     )
-    return scim_error.model_dump_json(), HTTPStatus.NOT_FOUND
 # -- resource-types-end --
 
 

doc/guides/_examples/integrations.py

@@ -25,9 +25,15 @@ def get_record(record_id):
     return records[record_id]
 
 
-def list_records():
-    """Return all stored records as a list."""
-    return list(records.values())
+def list_records(start=None, stop=None):
+    """Return a page of stored records and the total count.
+
+    :param start: 0-based start index.
+    :param stop: 0-based stop index (exclusive).
+    :return: A ``(total, page)`` tuple.
+    """
+    all_records = list(records.values())
+    return len(all_records), all_records[start:stop]
 
 
 def save_record(record):
@@ -73,14 +79,44 @@ def from_scim_user(scim_user):
 RESOURCE_MODELS = [User]
 
 
-def get_schemas():
-    """Return a :class:`~scim2_models.Schema` for every resource the server exposes."""
-    return [model.to_schema() for model in RESOURCE_MODELS]
+def get_schemas(start=None, stop=None):
+    """Return a page of :class:`~scim2_models.Schema` and the total count.
+
+    :param start: 0-based start index.
+    :param stop: 0-based stop index (exclusive).
+    :return: A ``(total, page)`` tuple.
+    """
+    all_schemas = [model.to_schema() for model in RESOURCE_MODELS]
+    return len(all_schemas), all_schemas[start:stop]
+
+
+def get_schema(schema_id):
+    """Return the :class:`~scim2_models.Schema` matching *schema_id*, or raise KeyError."""
+    for model in RESOURCE_MODELS:
+        schema = model.to_schema()
+        if schema.id == schema_id:
+            return schema
+    raise KeyError(schema_id)
+
+
+def get_resource_types(start=None, stop=None):
+    """Return a page of :class:`~scim2_models.ResourceType` and the total count.
+
+    :param start: 0-based start index.
+    :param stop: 0-based stop index (exclusive).
+    :return: A ``(total, page)`` tuple.
+    """
+    all_resource_types = [ResourceType.from_resource(model) for model in RESOURCE_MODELS]
+    return len(all_resource_types), all_resource_types[start:stop]
 
 
-def get_resource_types():
-    """Return a :class:`~scim2_models.ResourceType` for every resource the server exposes."""
-    return [ResourceType.from_resource(model) for model in RESOURCE_MODELS]
+def get_resource_type(resource_type_id):
+    """Return the :class:`~scim2_models.ResourceType` matching *resource_type_id*, or raise KeyError."""
+    for model in RESOURCE_MODELS:
+        rt = ResourceType.from_resource(model)
+        if rt.id == resource_type_id:
+            return rt
+    raise KeyError(resource_type_id)
 
 
 service_provider_config = ServiceProviderConfig(