from functools import partial
import json
from falcon import errors
from falcon import http_error
from falcon.media.base import BaseHandler
from falcon.media.base import TextBaseHandlerWS
[docs]class JSONHandler(BaseHandler):
"""JSON media handler.
This handler uses Python's standard :py:mod:`json` library by default, but
can be easily configured to use any of a number of third-party JSON
libraries, depending on your needs. For example, you can often
realize a significant performance boost under CPython by using an
alternative library. Good options in this respect include `orjson`,
`python-rapidjson`, and `mujson`.
This handler will raise a :class:`falcon.MediaNotFoundError` when attempting
to parse an empty body, or a :class:`falcon.MediaMalformedError`
if an error happens while parsing the body.
Note:
If you are deploying to PyPy, we recommend sticking with the standard
library's JSON implementation, since it will be faster in most cases
as compared to a third-party library.
Overriding the default JSON implementation is simply a matter of specifying
the desired ``dumps`` and ``loads`` functions::
import falcon
from falcon import media
import rapidjson
json_handler = media.JSONHandler(
dumps=rapidjson.dumps,
loads=rapidjson.loads,
)
extra_handlers = {
'application/json': json_handler,
}
app = falcon.App()
app.req_options.media_handlers.update(extra_handlers)
app.resp_options.media_handlers.update(extra_handlers)
By default, ``ensure_ascii`` is passed to the ``json.dumps`` function.
If you override the ``dumps`` function, you will need to explicitly set
``ensure_ascii`` to ``False`` in order to enable the serialization of
Unicode characters to UTF-8. This is easily done by using
:any:`functools.partial` to apply the desired keyword argument. In fact, you
can use this same technique to customize any option supported by the
``dumps`` and ``loads`` functions::
from functools import partial
from falcon import media
import rapidjson
json_handler = media.JSONHandler(
dumps=partial(
rapidjson.dumps,
ensure_ascii=False, sort_keys=True
),
)
Keyword Arguments:
dumps (func): Function to use when serializing JSON responses.
loads (func): Function to use when deserializing JSON requests.
"""
def __init__(self, dumps=None, loads=None):
self._dumps = dumps or partial(json.dumps, ensure_ascii=False)
self._loads = loads or json.loads
# PERF(kgriffs): Test dumps once up front so we can set the
# proper serialize implementation.
result = self._dumps({'message': 'Hello World'})
if isinstance(result, str):
self.serialize = self._serialize_s
self.serialize_async = self._serialize_async_s
else:
self.serialize = self._serialize_b
self.serialize_async = self._serialize_async_b
# NOTE(kgriffs): To be safe, only enable the optimized protocol when
# not subclassed.
if type(self) is JSONHandler:
self._serialize_sync = self.serialize
self._deserialize_sync = self._deserialize
def _deserialize(self, data):
if not data:
raise errors.MediaNotFoundError('JSON')
try:
return self._loads(data.decode())
except ValueError as err:
raise errors.MediaMalformedError('JSON') from err
def deserialize(self, stream, content_type, content_length):
return self._deserialize(stream.read())
async def deserialize_async(self, stream, content_type, content_length):
return self._deserialize(await stream.read())
# NOTE(kgriffs): Make content_type a kwarg to support the
# Request.render_body() shortcut optimization.
def _serialize_s(self, media, content_type=None) -> bytes:
return self._dumps(media).encode()
async def _serialize_async_s(self, media, content_type) -> bytes:
return self._dumps(media).encode()
def _serialize_b(self, media, content_type) -> bytes:
return self._dumps(media)
async def _serialize_async_b(self, media, content_type) -> bytes:
return self._dumps(media)
[docs]class JSONHandlerWS(TextBaseHandlerWS):
"""WebSocket media handler for de(serializing) JSON to/from TEXT payloads.
This handler uses Python's standard :py:mod:`json` library by default, but
can be easily configured to use any of a number of third-party JSON
libraries, depending on your needs. For example, you can often
realize a significant performance boost under CPython by using an
alternative library. Good options in this respect include `orjson`,
`python-rapidjson`, and `mujson`.
Note:
If you are deploying to PyPy, we recommend sticking with the standard
library's JSON implementation, since it will be faster in most cases
as compared to a third-party library.
Overriding the default JSON implementation is simply a matter of specifying
the desired ``dumps`` and ``loads`` functions::
import falcon
from falcon import media
import rapidjson
json_handler = media.JSONHandlerWS(
dumps=rapidjson.dumps,
loads=rapidjson.loads,
)
app = falcon.asgi.App()
app.ws_options.media_handlers[falcon.WebSocketPayloadType.TEXT] = json_handler
By default, ``ensure_ascii`` is passed to the ``json.dumps`` function.
If you override the ``dumps`` function, you will need to explicitly set
``ensure_ascii`` to ``False`` in order to enable the serialization of
Unicode characters to UTF-8. This is easily done by using
:any:`functools.partial` to apply the desired keyword argument. In fact, you
can use this same technique to customize any option supported by the
``dumps`` and ``loads`` functions::
from functools import partial
from falcon import media
import rapidjson
json_handler = media.JSONHandlerWS(
dumps=partial(
rapidjson.dumps,
ensure_ascii=False, sort_keys=True
),
)
Keyword Arguments:
dumps (func): Function to use when serializing JSON.
loads (func): Function to use when deserializing JSON.
"""
__slots__ = ['dumps', 'loads']
def __init__(self, dumps=None, loads=None):
self._dumps = dumps or partial(json.dumps, ensure_ascii=False)
self._loads = loads or json.loads
def serialize(self, media: object) -> str:
return self._dumps(media)
def deserialize(self, payload: str) -> object:
return self._loads(payload)
http_error._DEFAULT_JSON_HANDLER = _DEFAULT_JSON_HANDLER = JSONHandler() # type: ignore
