scim-aware model_validate_json

doc/changelog.rst

@@ -10,6 +10,9 @@ Added
   They look models up by schema, accept any sequence of :class:`~scim2_models.ScimObject` subclasses,
   and reflect the input types in the returned type.
 - :class:`~scim2_models.ScimObject` and ``AnyScimObject`` are exposed in the public API, so that downstream projects can annotate values that are either resources or messages.
+- :meth:`~scim2_models.BaseModel.model_validate_json` takes a ``scim_ctx`` parameter, like the
+  other validation and serialization methods, so JSON payloads can be validated without being
+  decoded first. :issue:`150`
 
 Changed
 ^^^^^^^

doc/tutorial.rst

@@ -55,6 +55,18 @@ Use Pydantic's :func:`~scim2_models.BaseModel.model_validate` method to parse an
     >>> user.meta.created  # doctest: +ELLIPSIS
     datetime.datetime(2010, 1, 23, 4, 56, 22, tzinfo=...)
 
+Payloads that have not been decoded yet can be handled by
+:func:`~scim2_models.BaseModel.model_validate_json`.
+Malformed JSON raises a :class:`~pydantic.ValidationError`, like any other invalid payload.
+
+.. code-block:: python
+
+    >>> import json
+
+    >>> user = User.model_validate_json(json.dumps(payload))
+    >>> user.user_name
+    'bjensen@example.com'
+
 
 Model serialization
 ===================
@@ -124,6 +136,8 @@ fields with unexpected values will raise :class:`~pydantic.ValidationError`:
     ... except ValidationError:
     ...    obj = Error(...)
 
+:meth:`~scim2_models.BaseModel.model_validate_json` takes the same :paramref:`~scim2_models.BaseModel.model_validate_json.scim_ctx` parameter.
+
 Context annotations
 ===================
 

scim2_models/base.py

@@ -676,6 +676,18 @@ def _scim_response_serializer(
             ):
                 del serialized[alias]
 
+    @classmethod
+    def _prepare_model_validate(
+        cls,
+        scim_ctx: Context | None = Context.DEFAULT,
+        original: Optional["BaseModel"] = None,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        context = kwargs.setdefault("context", {})
+        context.setdefault("scim", scim_ctx)
+        context.setdefault("original", original)
+        return kwargs
+
     @classmethod
     def model_validate(
         cls,
@@ -704,11 +716,25 @@ def model_validate(
                 stacklevel=2,
             )
 
-        context = kwargs.setdefault("context", {})
-        context.setdefault("scim", scim_ctx)
-        context.setdefault("original", original)
+        validate_kwargs = cls._prepare_model_validate(scim_ctx, original, **kwargs)
+        return super().model_validate(*args, **validate_kwargs)
+
+    @classmethod
+    def model_validate_json(
+        cls,
+        *args: Any,
+        scim_ctx: Context | None = Context.DEFAULT,
+        **kwargs: Any,
+    ) -> Self:
+        """Validate SCIM JSON payloads and generate model representation by using Pydantic :meth:`~pydantic.BaseModel.model_validate_json`.
 
-        return super().model_validate(*args, **kwargs)
+        Malformed JSON payloads raise a :class:`~pydantic.ValidationError`, like
+        any other SCIM validation failure.
+
+        :param scim_ctx: The SCIM :class:`~scim2_models.Context` in which the validation happens.
+        """
+        validate_kwargs = cls._prepare_model_validate(scim_ctx, **kwargs)
+        return super().model_validate_json(*args, **validate_kwargs)
 
     def _prepare_model_dump(
         self,

tests/test_model_validation.py

@@ -746,3 +746,49 @@ def test_validate_query_and_search_request_necessity(context):
         id="x",
         optional="x",
     )
+
+
+def test_validate_json_payload():
+    """JSON payloads are validated like already decoded payloads."""
+    assert MutResource.model_validate_json('{"readWrite": "x"}') == MutResource(
+        schemas=["org:example:MutResource"],
+        readWrite="x",
+    )
+
+
+def test_validate_json_bytes_payload():
+    """JSON payloads can be passed as bytes."""
+    assert MutResource.model_validate_json(b'{"readWrite": "x"}') == MutResource(
+        schemas=["org:example:MutResource"],
+        readWrite="x",
+    )
+
+
+def test_validate_json_applies_the_scim_context():
+    """The SCIM context drives the validation of JSON payloads."""
+    with pytest.raises(
+        ValidationError,
+        match="Field 'write_only' has mutability 'writeOnly' but this in not valid in resource query request context",
+    ):
+        MutResource.model_validate_json(
+            '{"writeOnly": "x"}',
+            scim_ctx=Context.RESOURCE_QUERY_REQUEST,
+        )
+
+
+def test_validate_json_with_an_explicit_validation_context():
+    """An explicit Pydantic validation context takes precedence over the SCIM context."""
+    with pytest.raises(
+        ValidationError,
+        match="Field 'write_only' has mutability 'writeOnly' but this in not valid in resource query request context",
+    ):
+        MutResource.model_validate_json(
+            '{"writeOnly": "x"}',
+            context={"scim": Context.RESOURCE_QUERY_REQUEST},
+        )
+
+
+def test_validate_json_rejects_malformed_payloads():
+    """Malformed JSON payloads raise a ValidationError like any invalid payload."""
+    with pytest.raises(ValidationError, match="Invalid JSON"):
+        MutResource.model_validate_json("{invalid")

tests/test_models.py

@@ -1,6 +1,8 @@
+import json
 import os
 
 import pytest
+from pydantic import ValidationError
 
 from scim2_models import BulkRequest
 from scim2_models import BulkResponse
@@ -19,28 +21,32 @@
 from scim2_models import get_model_by_schema
 
 
+def _error_summary(exc: ValidationError) -> list[tuple[str, tuple]]:
+    return [(error["type"], error["loc"]) for error in exc.errors()]
+
+
+SAMPLE_MODELS = {
+    "user": User,
+    "enterprise_user": User[EnterpriseUser],
+    "group": Group,
+    "schema": Schema,
+    "resource_type": ResourceType,
+    "service_provider_configuration": ServiceProviderConfig,
+    "list_response": ListResponse[User[EnterpriseUser] | Group | Schema | ResourceType],
+    "patch_op": PatchOp[User],
+    "bulk_request": BulkRequest,
+    "bulk_response": BulkResponse,
+    "search_request": SearchRequest,
+    "error": Error,
+}
+
+
 def test_parse_and_serialize_examples(load_sample):
     samples = list(os.walk("samples"))[0][2]
-    models = {
-        "user": User,
-        "enterprise_user": User[EnterpriseUser],
-        "group": Group,
-        "schema": Schema,
-        "resource_type": ResourceType,
-        "service_provider_configuration": ServiceProviderConfig,
-        "list_response": ListResponse[
-            User[EnterpriseUser] | Group | Schema | ResourceType
-        ],
-        "patch_op": PatchOp[User],
-        "bulk_request": BulkRequest,
-        "bulk_response": BulkResponse,
-        "search_request": SearchRequest,
-        "error": Error,
-    }
 
     for sample in samples:
         model_name = sample.replace(".json", "").split("-")[2]
-        model = models[model_name]
+        model = SAMPLE_MODELS[model_name]
 
         skipped = [
             # resources without schemas are not yet supported
@@ -76,6 +82,32 @@ def test_parse_and_serialize_examples(load_sample):
         assert obj.model_dump(exclude_unset=True) == payload
 
 
+def test_parse_json_and_decoded_examples(load_sample):
+    """JSON payloads and already decoded payloads are validated the same way."""
+    samples = list(os.walk("samples"))[0][2]
+
+    for sample in samples:
+        model_name = sample.replace(".json", "").split("-")[2]
+        model = SAMPLE_MODELS[model_name]
+
+        payload = load_sample(sample)
+        raw = json.dumps(payload)
+
+        try:
+            obj = model.model_validate(payload)
+        except ValidationError as exc:
+            with pytest.raises(ValidationError) as json_exc:
+                model.model_validate_json(raw)
+            assert _error_summary(json_exc.value) == _error_summary(exc)
+            continue
+
+        json_obj = model.model_validate_json(raw)
+        assert obj == json_obj
+        assert obj.model_dump(exclude_unset=True) == json_obj.model_dump(
+            exclude_unset=True
+        )
+
+
 def test_get_model_by_schema():
     resource_types = [Group, User[EnterpriseUser]]
     assert (