feat: Pydantic field validation (#101)

* Require pydantic
* Add tests for Pydantic field validation / coercion
* Add Pydantic validation / coercion
* Autoformat
* CHANGELOG entry
* Update docstring
* Add nested Pydantic testing

Author: bgrant

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 # [unreleased] - XXXX-XX-XX
 ### Added
 - [PR 100](https://github.com/salesforce/django-declarative-apis/pull/100) Improve `errors.py`
+- [PR 101](https://github.com/salesforce/django-declarative-apis/pull/101) Allow Pydantic models as field types
 
 # [0.23.1] - 2022-05-17
 ### Fixed

django_declarative_apis/machinery/attributes.py

@@ -13,6 +13,7 @@
 import time
 
 from django.db import models as django_models
+import pydantic
 
 from . import errors
 from . import tasks
@@ -112,7 +113,9 @@ def documentation(self):
 class TypedEndpointAttributeMixin:
     def __init__(self, *args, **kwargs):
         self.field_type = kwargs.pop("type", str)
-        if self.field_type not in RequestField.VALID_FIELD_TYPES:
+        if not any(
+            issubclass(self.field_type, t) for t in RequestField.VALID_FIELD_TYPES
+        ):
             raise NotImplementedError(
                 "Request fields of type {0} not supported".format(
                     self.field_type.__name__
@@ -124,6 +127,8 @@ def coerce_value_to_type(self, raw_value):
         try:
             if self.field_type == bool and not isinstance(raw_value, self.field_type):
                 return "rue" in raw_value
+            elif issubclass(self.field_type, pydantic.BaseModel):
+                return self.field_type.parse_obj(raw_value)
             else:
                 if isinstance(raw_value, collections.abc.Iterable) and not isinstance(
                     raw_value, (str, dict)
@@ -201,8 +206,8 @@ class RequestField(TypedEndpointAttributeMixin, RequestProperty):
     """Endpoint properties are called fields. Fields can be simple types such as int,
     or they can be used as a decorator on a function.
 
-    **Valid field types:** :code:`int`, :code:`bool`, :code:`float`, :code:`str`,
-    :code:`dict`, :code:`complex`
+    **Valid field types:** A subclass of :code:`int`, :code:`bool`, :code:`float`,
+    :code:`str`, :code:`dict`, :code:`complex`, :code:`pydantic.BaseModel`
 
     **Example**
 
@@ -258,7 +263,7 @@ class FooDefinition(EndpointDefinition):
 
     """
 
-    VALID_FIELD_TYPES = (bool, int, float, complex, str, dict)
+    VALID_FIELD_TYPES = (bool, int, float, complex, str, dict, pydantic.BaseModel)
 
     def __init__(self, *args, **kwargs):
         self.default_value = kwargs.pop("default", None)

requirements.txt

@@ -4,3 +4,4 @@ cryptography>=2.0,<=3.4.8
 decorator==4.0.11
 django-dirtyfields>=1.2.1
 oauthlib[signedtoken,rsa]>=2.0.6,<3.1.0
+pydantic>=1.8

tests/tests.py

@@ -76,3 +76,104 @@ def test_run_deferred_task(self):
             "/simple", consumer=self.consumer, expected_status_code=HTTPStatus.OK
         )
         self.assertTrue(cache.get("deferred_task_called"))
+
+    def test_dict_field_endpoint(self):
+        good_dict = {
+            "length": 11,
+            "description": "This is a description",
+            "timestamp": "2022-10-24T00:00:00",
+            "words": ["foo", "bar", "baz", "quux"],
+        }
+        test_data = [
+            (good_dict, HTTPStatus.OK, "good dict"),
+            ({}, HTTPStatus.OK, "empty_dict"),
+            (list(good_dict), HTTPStatus.BAD_REQUEST, "list"),
+            ("a string", HTTPStatus.BAD_REQUEST, "string"),
+            (1337, HTTPStatus.BAD_REQUEST, "int"),
+        ]
+
+        for dct, expected_status, message in test_data:
+            data = {"dict_type_field": dct}
+            with self.subTest(message):
+                response = self.client.post(
+                    "/dictfield",
+                    consumer=self.consumer,
+                    data=data,
+                    expected_status_code=expected_status,
+                    content_type="application/json",
+                )
+                if expected_status == HTTPStatus.OK:
+                    self.assertDictEqual(json.loads(response.content), data)
+
+    def test_pydantic_field_endpoint(self):
+        good_dict = {
+            "length": 11,
+            "description": "This is a description",
+            "timestamp": "2022-10-24T00:00:00",
+            "words": ["foo", "bar", "baz", "quux"],
+        }
+        test_data = [
+            (good_dict, HTTPStatus.OK, "no errors"),
+            ({**good_dict, "length": "eleven"}, HTTPStatus.BAD_REQUEST, "bad length"),
+            (
+                {**good_dict, "description": ["one", "two"]},
+                HTTPStatus.BAD_REQUEST,
+                "bad description",
+            ),
+            (
+                {**good_dict, "timestamp": "2022-10-24T99:99:99"},
+                HTTPStatus.BAD_REQUEST,
+                "bad timestamp",
+            ),
+            (
+                {**good_dict, "words": "foo bar baz quux"},
+                HTTPStatus.BAD_REQUEST,
+                "bad words",
+            ),
+        ]
+
+        for dct, expected_status, message in test_data:
+            data = {"pydantic_type_field": dct}
+            with self.subTest(message):
+                response = self.client.post(
+                    "/pydanticfield",
+                    consumer=self.consumer,
+                    data=data,
+                    expected_status_code=expected_status,
+                    content_type="application/json",
+                )
+                if expected_status == HTTPStatus.OK:
+                    self.assertDictEqual(json.loads(response.content), data)
+
+    def test_nested_pydantic_field_endpoint(self):
+        good_dict = {"b": "hello", "c": {"a": "world"}}
+        test_data = [
+            (good_dict, HTTPStatus.OK, "no errors"),
+            ({**good_dict, "b": list("abc")}, HTTPStatus.BAD_REQUEST, "bad b"),
+            ({**good_dict, "c": 11}, HTTPStatus.BAD_REQUEST, "bad c"),
+            ({**good_dict, "c": {"a": list("abc")}}, HTTPStatus.BAD_REQUEST, "bad a"),
+            ({**good_dict, "c": {}}, HTTPStatus.BAD_REQUEST, "missing a"),
+            (
+                {k: v for (k, v) in good_dict.items() if k != "b"},
+                HTTPStatus.BAD_REQUEST,
+                "missing b",
+            ),
+            (
+                {k: v for (k, v) in good_dict.items() if k != "c"},
+                HTTPStatus.BAD_REQUEST,
+                "missing c",
+            ),
+        ]
+
+        for dct, expected_status, message in test_data:
+            data = {"nested_pydantic_type_field": dct}
+            with self.subTest(message):
+                response = self.client.post(
+                    "/nestedpydanticfield",
+                    consumer=self.consumer,
+                    data=data,
+                    expected_status_code=expected_status,
+                    content_type="application/json",
+                )
+                if expected_status == HTTPStatus.OK:
+                    self.assertDictEqual(json.loads(response.content), data)

tests/urls.py

@@ -14,6 +14,14 @@
 UUID4_REGEX = r"[0-9a-fA-F]{8}-([0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}"
 
 urlpatterns = [
-    re_path(r"^simple", resource_adapter(get=views.SimpleEndpointDefinition)),
-    re_path(r"^dict", resource_adapter(get=views.DictEndpointDefinition)),
+    re_path(r"^simple$", resource_adapter(get=views.SimpleEndpointDefinition)),
+    re_path(r"^dict$", resource_adapter(get=views.DictEndpointDefinition)),
+    re_path(r"^dictfield$", resource_adapter(post=views.DictFieldEndpointDefinition)),
+    re_path(
+        r"^pydanticfield$", resource_adapter(post=views.PydanticFieldEndpointDefinition)
+    ),
+    re_path(
+        r"^nestedpydanticfield$",
+        resource_adapter(post=views.NestedPydanticFieldEndpointDefinition),
+    ),
 ]

tests/views.py

@@ -5,14 +5,20 @@
 # For full license text, see the LICENSE file in the repo root or https://opensource.org/licenses/BSD-3-Clause
 #
 
+import datetime
+import json
+from typing import List
+
+from django.core.cache import cache
+
+import pydantic
+
 from django_declarative_apis.machinery import (
     EndpointDefinition,
     field,
     deferrable_task,
     endpoint_resource,
 )
-from django.core.cache import cache
-
 from tests.models import TestModel
 
 
@@ -44,3 +50,59 @@ def is_authorized(self):
     def resource(self):
         inst = TestModel.objects.create(int_field=1)
         return {"test": inst, "deep_test": {"test": inst}}
+
+
+class DictFieldEndpointDefinition(EndpointDefinition):
+    def is_authorized(self):
+        return True
+
+    dict_type_field = field(type=dict, required=True)
+
+    @endpoint_resource(type=dict)
+    def resource(self):
+        return {
+            "dict_type_field": self.dict_type_field,
+        }
+
+
+class PydanticFieldEndpointDefinition(EndpointDefinition):
+    def is_authorized(self):
+        return True
+
+    class _TestData(pydantic.BaseModel):
+        length: int
+        description: str
+        timestamp: datetime.datetime
+        words: List[str]
+
+    pydantic_type_field = field(type=_TestData, required=True)
+
+    @endpoint_resource(type=dict)
+    def resource(self):
+        return {
+            "pydantic_type_field": json.loads(self.pydantic_type_field.json()),
+        }
+
+
+class _ModelA(pydantic.BaseModel):
+    a: str
+
+
+class _ModelB(pydantic.BaseModel):
+    b: str
+    c: _ModelA
+
+
+class NestedPydanticFieldEndpointDefinition(EndpointDefinition):
+    def is_authorized(self):
+        return True
+
+    nested_pydantic_type_field = field(type=_ModelB, required=True)
+
+    @endpoint_resource(type=dict)
+    def resource(self):
+        return {
+            "nested_pydantic_type_field": json.loads(
+                self.nested_pydantic_type_field.json()
+            ),
+        }