refactor: validate bulk requests through scim2-models

Author: azmeuk

scim2_server/bulk.py

@@ -1,55 +1,71 @@
 from collections.abc import Callable
 from typing import Any
 
+from pydantic import BaseModel
+from scim2_models import BulkOperation
 from scim2_models import InvalidValueException
 from scim2_models import Resource
 from werkzeug.exceptions import Conflict
 
 BULK_ID_PREFIX = "bulkId:"
 
-Resolver = Callable[[Any], Any]
-"""Replaces the bulkId references of a raw operation."""
+Resolver = Callable[[BulkOperation], BulkOperation]
+"""Replaces the bulkId references of an operation."""
 
-OperationRunner = Callable[[Any, Resolver], tuple[dict[str, Any], Resource | None]]
-"""Applies a raw operation once resolved, and returns its outcome and the resource it acted on."""
-
-
-def raw_attribute(payload: Any, name: str) -> Any:
-    """Return an attribute of a raw payload, whose names are case insensitive."""
-    if not isinstance(payload, dict):
-        return None
-    return next(
-        (value for key, value in payload.items() if key.casefold() == name.casefold()),
-        None,
-    )
+OperationRunner = Callable[
+    [BulkOperation, Resolver], tuple[dict[str, Any], Resource | None]
+]
+"""Applies an operation once resolved, and returns its outcome and the resource it acted on."""
 
 
 def replace_bulk_ids(value: Any, replace: Callable[[str], str]) -> Any:
-    """Replace every "bulkId:" reference of a raw value."""
+    """Replace every "bulkId:" reference of a value.
+
+    A value without reference is returned as it is. Models are copied with
+    only the changed fields, so the fields the client set stay the same.
+    """
     if isinstance(value, str) and value.startswith(BULK_ID_PREFIX):
         return replace(value.removeprefix(BULK_ID_PREFIX))
+
     if isinstance(value, list):
-        return [replace_bulk_ids(item, replace) for item in value]
+        items = [replace_bulk_ids(item, replace) for item in value]
+        changed = any(new is not old for new, old in zip(items, value, strict=True))
+        return items if changed else value
+
     if isinstance(value, dict):
-        return {key: replace_bulk_ids(item, replace) for key, item in value.items()}
+        entries = {key: replace_bulk_ids(item, replace) for key, item in value.items()}
+        changed = any(entries[key] is not item for key, item in value.items())
+        return entries if changed else value
+
+    if isinstance(value, BaseModel):
+        updates = {}
+        for name in type(value).model_fields:
+            field = getattr(value, name)
+            replaced = replace_bulk_ids(field, replace)
+            if replaced is not field:
+                updates[name] = replaced
+        return value.model_copy(update=updates) if updates else value
+
     return value
 
 
-def resolve_operation(payload: Any, replace: Callable[[str], str]) -> Any:
-    """Replace the "bulkId:" references of the path and the data of a raw bulk operation."""
-    if not isinstance(payload, dict):
-        return payload
+def resolve_operation(
+    operation: BulkOperation, replace: Callable[[str], str]
+) -> BulkOperation:
+    """Replace the "bulkId:" references of the path and the data of a bulk operation."""
+    updates: dict[str, Any] = {}
+    if operation.path is not None:
+        path = "/".join(
+            replace_bulk_ids(segment, replace) for segment in operation.path.split("/")
+        )
+        if path != operation.path:
+            updates["path"] = path
 
-    resolved = {}
-    for key, value in payload.items():
-        if key.casefold() == "path" and isinstance(value, str):
-            value = "/".join(
-                replace_bulk_ids(segment, replace) for segment in value.split("/")
-            )
-        elif key.casefold() == "data":
-            value = replace_bulk_ids(value, replace)
-        resolved[key] = value
-    return resolved
+    data = replace_bulk_ids(operation.data, replace)
+    if data is not operation.data:
+        updates["data"] = data
+
+    return operation.model_copy(update=updates) if updates else operation
 
 
 class BulkJob:
@@ -63,7 +79,7 @@ class BulkJob:
 
     def __init__(
         self,
-        operations: list[Any],
+        operations: list[BulkOperation],
         fail_on_errors: int | None,
         run: OperationRunner,
     ):
@@ -76,10 +92,12 @@ def __init__(
         self.errors = 0
 
         self.creations: dict[str, int] = {}
-        for index, payload in enumerate(operations):
-            bulk_id = raw_attribute(payload, "bulkId")
-            if raw_attribute(payload, "method") == "POST" and isinstance(bulk_id, str):
-                self.creations.setdefault(bulk_id, index)
+        for index, operation in enumerate(operations):
+            if (
+                operation.method == BulkOperation.Method.post
+                and operation.bulk_id is not None
+            ):
+                self.creations.setdefault(operation.bulk_id, index)
 
     @property
     def stopped(self) -> bool:
@@ -105,15 +123,15 @@ def run_operation(self, index: int) -> None:
         if index in self.results or index in self.running or self.stopped:
             return
 
-        payload = self.operations[index]
+        operation = self.operations[index]
         self.running.add(index)
-        for bulk_id in self.references(payload):
+        for bulk_id in self.references(operation):
             if bulk_id in self.creations:
                 self.run_operation(self.creations[bulk_id])
 
         if not self.stopped:
             result, resource = self.run_resolved(
-                payload, lambda payload: self.resolve(index, payload)
+                operation, lambda operation: self.resolve(index, operation)
             )
             self.results[index] = result
             if result["status"] >= 400:
@@ -127,31 +145,30 @@ def is_creation(self, index: int, bulk_id: str | None) -> bool:
         return bulk_id is not None and self.creations.get(bulk_id) == index
 
     @staticmethod
-    def references(payload: Any) -> list[str]:
+    def references(operation: BulkOperation) -> list[str]:
         """Return the bulkIds an operation references."""
         bulk_ids: list[str] = []
 
         def collect(bulk_id: str) -> str:
             bulk_ids.append(bulk_id)
             return bulk_id
 
-        resolve_operation(payload, collect)
+        resolve_operation(operation, collect)
         return bulk_ids
 
-    def resolve(self, index: int, payload: Any) -> Any:
+    def resolve(self, index: int, operation: BulkOperation) -> BulkOperation:
         """Replace the bulkId references of an operation with the identifiers of the created resources.
 
         :raises Conflict: When a referenced resource was not created, as
             RFC 7644 §3.7.1 allows for circular references.
         """
-        bulk_id = raw_attribute(payload, "bulkId")
         if (
-            raw_attribute(payload, "method") == "POST"
-            and isinstance(bulk_id, str)
-            and not self.is_creation(index, bulk_id)
+            operation.method == BulkOperation.Method.post
+            and operation.bulk_id is not None
+            and not self.is_creation(index, operation.bulk_id)
         ):
             raise InvalidValueException(
-                detail=f"The bulkId {bulk_id} is not unique in the request"
+                detail=f"The bulkId {operation.bulk_id} is not unique in the request"
             )
 
         def replace(bulk_id: str) -> str:
@@ -161,4 +178,4 @@ def replace(bulk_id: str) -> str:
                 raise Conflict(f"The bulkId {bulk_id} is part of a circular reference")
             raise Conflict(f"No resource was created with the bulkId {bulk_id}")
 
-        return resolve_operation(payload, replace)
+        return resolve_operation(operation, replace)

scim2_server/provider.py

@@ -47,7 +47,6 @@
 from scim2_server.backend import Backend
 from scim2_server.bulk import BulkJob
 from scim2_server.bulk import Resolver
-from scim2_server.bulk import raw_attribute
 from scim2_server.utils import load_default_service_provider_config
 
 SEARCH_REQUEST_PARAMETERS = (
@@ -516,7 +515,10 @@ def call_bulk(self, request: Request, **kwargs) -> Response:
                 f"The payload exceeds the maxPayloadSize ({bulk.max_payload_size} bytes)"
             )
 
-        bulk_request, operations = self.read_bulk_request(request.json)
+        bulk_request = BulkRequest[Union[tuple(self.get_models())]].model_validate(  # noqa: UP007
+            request.json, scim_ctx=Context.BULK_REQUEST
+        )
+        operations = cast(list[BulkOperation], bulk_request.operations)
         if bulk.max_operations is not None and len(operations) > bulk.max_operations:
             raise RequestEntityTooLarge(
                 f"The number of operations exceeds the maxOperations ({bulk.max_operations})"
@@ -525,59 +527,48 @@ def call_bulk(self, request: Request, **kwargs) -> Response:
         results = BulkJob(
             operations,
             bulk_request.fail_on_errors,
-            lambda payload, resolve: self.run_bulk_operation(request, payload, resolve),
+            lambda operation, resolve: self.run_bulk_operation(
+                request, operation, resolve
+            ),
         ).run()
         return self.make_response(
             BulkResponse[Union[tuple(self.get_models())]](  # noqa: UP007
                 operations=results
             ).model_dump(scim_ctx=Context.BULK_RESPONSE)
         )
 
-    def read_bulk_request(self, payload: Any) -> tuple[BulkRequest, list[Any]]:
-        """Validate the envelope of a bulk request, and return it with its raw operations.
-
-        Each operation is validated on its own, so an invalid operation only
-        fails itself (RFC 7644 §3.7.3).
-        """
-        key = (
-            next((key for key in payload if key.casefold() == "operations"), None)
-            if isinstance(payload, dict)
-            else None
-        )
-        operations = payload[key] if key else None
-        envelope = {**payload, key: []} if isinstance(operations, list) else payload
-        bulk_request = BulkRequest[Union[tuple(self.get_models())]].model_validate(  # noqa: UP007
-            envelope, scim_ctx=Context.BULK_REQUEST
-        )
-        return bulk_request, cast(list[Any], operations)
-
     def run_bulk_operation(
-        self, request: Request, payload: Any, resolve: Resolver
+        self, request: Request, operation: BulkOperation, resolve: Resolver
     ) -> tuple[dict[str, Any], Resource | None]:
         """Apply one operation of a bulk job.
 
+        An operation that failed its validation keeps its error, once its
+        references are resolved to locate it.
+
         :return: The outcome of the operation, and the resource it created or updated.
         """
-        method = raw_attribute(payload, "method")
-        bulk_id = raw_attribute(payload, "bulkId")
         result: dict[str, Any] = {
-            "method": method
-            if method in [member.value for member in BulkOperation.Method]
-            else None,
-            "bulk_id": bulk_id if isinstance(bulk_id, str) else None,
+            "method": operation.method,
+            "bulk_id": operation.bulk_id,
         }
 
         try:
-            payload = resolve(payload)
-            resource_type, resource_id = self.get_bulk_target(payload)
-            if resource_id:
+            operation = resolve(operation)
+            resource_type = self.get_resource_type_by_endpoint(operation.endpoint or "")
+            if resource_type is not None and operation.resource_id:
                 result["location"] = urljoin(
-                    request.url, f"{resource_type.endpoint.strip('/')}/{resource_id}"
+                    request.url,
+                    f"{resource_type.endpoint.strip('/')}/{operation.resource_id}",
                 )
-            operation = BulkOperation[self.get_model(resource_type)].model_validate(
-                payload, scim_ctx=Context.BULK_REQUEST
+            if isinstance(operation.response, Error):
+                return {
+                    **result,
+                    "status": operation.status,
+                    "response": operation.response,
+                }, None
+            resource = self.apply_bulk_operation(
+                cast(ResourceType, resource_type), operation
             )
-            resource = self.apply_bulk_operation(resource_type, resource_id, operation)
         except Exception as exception:
             error = self.error_from(exception)
             return {**result, "status": error.status, "response": error}, None
@@ -591,34 +582,15 @@ def run_bulk_operation(
         result["version"] = resource.meta.version
         return result, resource
 
-    def get_bulk_target(self, payload: Any) -> tuple[ResourceType, str]:
-        """Return the resource type and the resource identifier of a bulk operation path.
-
-        :raises NotFound: When the path does not start with a resource type endpoint.
-        """
-        path = raw_attribute(payload, "path")
-        if not isinstance(path, str):
-            raise InvalidValueException(
-                detail="path is required for request operations"
-            )
-
-        endpoint, _, resource_id = path.lstrip("/").partition("/")
-        resource_type = self.get_resource_type_by_endpoint(endpoint)
-        if resource_type is None:
-            raise NotFound
-        return resource_type, resource_id
-
     def apply_bulk_operation(
-        self,
-        resource_type: ResourceType,
-        resource_id: str,
-        operation: BulkOperation,
+        self, resource_type: ResourceType, operation: BulkOperation
     ) -> Resource | None:
         """Apply a validated bulk operation, and return the resource it acted on.
 
         The data of the operation is already validated, and the resource
         operations take it as it is.
         """
+        resource_id = operation.resource_id
         if (operation.method == BulkOperation.Method.post) == bool(resource_id):
             raise InvalidValueException(
                 detail="A POST path must target a resource type endpoint, other methods a resource"