fix: Partial payload patch (#9)

* fix: Partial payload patch

* chore: Optimize import

* fix: Patch schema did not have original field's metadata and validations

django_ninja_crudl/__init__.py

@@ -1,7 +1,9 @@
 """Super schema packages."""
 
 from django2pydantic import Infer, ModelFields  # hoisting/bubble up
+
 from django_ninja_crudl.crudl import Crudl, CrudlApiBaseMeta
+from django_ninja_crudl.patch_dict import PatchDict
 from django_ninja_crudl.permissions import BasePermission
 from django_ninja_crudl.types import (
     ObjectlessActions,
@@ -17,6 +19,7 @@
     "CrudlApiBaseMeta",
     "ObjectlessActions",
     "WithObjectActions",
+    "PatchDict",
     "PathArgs",
     "RequestDetails",
     "BasePermission",

django_ninja_crudl/crudl.py

@@ -18,7 +18,6 @@
     OneToOneRel,
 )
 from django.http import HttpRequest, HttpResponse
-from ninja import PatchDict
 from ninja_extra import (
     ControllerBase,
     api_controller,
@@ -48,6 +47,7 @@
     UnprocessableEntity422Schema,
 )
 from django_ninja_crudl.model_utils import get_pydantic_fields
+from django_ninja_crudl.patch_dict import PatchDict
 from django_ninja_crudl.permissions import BasePermission
 from django_ninja_crudl.types import PathArgs, RequestDetails
 from django_ninja_crudl.utils import add_function_arguments, validating_manager
@@ -126,6 +126,7 @@ def __new__(cls, name: str, bases: tuple[type, ...], dct: dict[str, Any]) -> typ
 
         model_class: type[Model] = meta.model_class
 
+
         api_meta = getattr(model_class, "CrudlApiMeta", meta)
         if api_meta is None:
             msg = f"CrudlApiMeta is required for model '{name}' or in the model itself."
@@ -595,14 +596,13 @@ def update(
                     path=update_path,
                     operation_id=patch_operation_id,
                     response={
-                        status.HTTP_200_OK: UpdateSchema,  # pyright: ignore[reportPossiblyUnboundVariable]
+                        status.HTTP_200_OK: PartialUpdateSchema,  # pyright: ignore[reportPossiblyUnboundVariable]
                         status.HTTP_401_UNAUTHORIZED: Unauthorized401Schema,
                         status.HTTP_403_FORBIDDEN: Forbidden403Schema,
                         status.HTTP_404_NOT_FOUND: ResourceNotFound404Schema,
                         status.HTTP_422_UNPROCESSABLE_ENTITY: UnprocessableEntity422Schema,
                         status.HTTP_503_SERVICE_UNAVAILABLE: ServiceUnavailable503Schema,
                     },
-                    exclude_unset=True,
                     by_alias=True,
                 )
                 @transaction.atomic

django_ninja_crudl/patch_dict.py

@@ -0,0 +1,51 @@
+from copy import deepcopy
+from typing import TYPE_CHECKING, Annotated, Any
+
+from ninja import Body
+from ninja.utils import is_optional_type
+from pydantic_core import core_schema
+
+
+class ModelToDict(dict):
+    _wrapped_model: Any = None
+    _wrapped_model_dump_params: dict[str, Any] = {}
+
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _source: Any, _handler: Any) -> Any:
+        return core_schema.no_info_after_validator_function(
+            cls._validate,
+            cls._wrapped_model.__pydantic_core_schema__,
+        )
+
+    @classmethod
+    def _validate(cls, input_value: Any) -> Any:
+        return input_value.model_dump(**cls._wrapped_model_dump_params)
+
+
+def create_patch_schema(schema_cls: type[Any]) -> type[ModelToDict]:
+    # Turn required fields into optional by assigning a default None value
+    schema_cls_copy = deepcopy(schema_cls)
+    for f in schema_cls_copy.__pydantic_fields__.values():
+        t = f.annotation
+        if not is_optional_type(t):
+            f.default = None
+    # The cloned schema should be recreated for the changes to take effect
+    OptionalSchema = type(f"{schema_cls.__name__}Patch", (schema_cls,), {})
+
+    class OptionalDictSchema(ModelToDict):
+        _wrapped_model = OptionalSchema
+        _wrapped_model_dump_params = {"exclude_unset": True}
+
+    return OptionalDictSchema
+
+
+class PatchDictUtil:
+    def __getitem__(self, schema_cls: Any) -> Any:
+        new_cls = create_patch_schema(schema_cls)
+        return Body[new_cls]  # type: ignore
+
+
+if TYPE_CHECKING:  # pragma: nocover
+    PatchDict = Annotated[dict, "<PatchDict>"]
+else:
+    PatchDict = PatchDictUtil()