Add custom loads function

sanic/app.py

@@ -173,6 +173,7 @@ def __init__(
         configure_logging: bool = True,
         register: Optional[bool] = None,
         dumps: Optional[Callable[..., AnyStr]] = None,
+        loads: Optional[Callable[[AnyStr], Optional[Dict]]] = None,
     ) -> None:
         super().__init__(name=name)
 
@@ -237,6 +238,8 @@ def __init__(
 
         if dumps:
             BaseHTTPResponse._dumps = dumps  # type: ignore
+        if loads:
+            Request._loads = loads  # type: ignore
 
     @property
     def loop(self):

sanic/request.py

@@ -118,6 +118,8 @@ class Request:
         "version",
     )
 
+    _loads = json_loads
+
     def __init__(
         self,
         url_bytes: bytes,
@@ -372,8 +374,11 @@ def json(self):
 
         return self.parsed_json
 
-    def load_json(self, loads=json_loads):
+    def load_json(self, loads=None):
         try:
+            if not loads:
+                loads = self.__class__._loads
+
             self.parsed_json = loads(self.body)
         except Exception:
             if not self.body:

tests/test_json_decoding.py

@@ -0,0 +1,53 @@
+from json import loads as sloads
+
+import pytest
+
+
+try:
+    from ujson import loads as uloads
+
+    NO_UJSON = False
+    DEFAULT_LOADS = uloads
+except ModuleNotFoundError:
+    NO_UJSON = True
+    DEFAULT_LOADS = sloads
+
+from sanic import Request, Sanic, json
+
+
+@pytest.fixture(autouse=True)
+def default_back_to_ujson():
+    yield
+    Request._loads = DEFAULT_LOADS
+
+
+def test_change_decoder():
+    Sanic("Test", loads=sloads)
+    assert Request._loads == sloads
+
+
+def test_change_decoder_to_some_custom():
+    def my_custom_decoder(some_str: str):
+        print ("decoding")
+        dict =  sloads(some_str)
+        dict["some_key"] = "new_value"
+        return dict
+
+    app = Sanic("Test", loads=my_custom_decoder)
+    assert Request._loads == my_custom_decoder
+
+    req_body = {'some_key':'some_value'}
+
+    @app.post("/test")
+    def handler(request):
+        new_json = request.json
+        return json(new_json)
+
+    req, res = app.test_client.post("/test", json=req_body)
+    assert sloads(res.body) == {'some_key':'new_value'}
+
+
+@pytest.mark.skipif(NO_UJSON is True, reason="ujson not installed")
+def test_default_decoder():
+    Sanic("Test")
+    assert Request._loads == uloads