falcon.media.json 源代码

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


[文档]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. .. rubric:: Custom JSON library You can replace the default JSON handler by using a custom JSON library (see also: :ref:`custom_media_handlers`). 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) .. rubric:: Custom serialization parameters Even if you decide to stick with the stdlib's :any:`json.dumps` and :any:`json.loads`, you can wrap them using :any:`functools.partial` to provide custom serialization or deserialization parameters supported by the ``dumps`` and ``loads`` functions, respectively (see also: :ref:`prettifying-json-responses`):: import falcon from falcon import media from functools import partial json_handler = media.JSONHandler( dumps=partial( json.dumps, default=str, sort_keys=True, ), ) 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. As also demonstrated in the previous paragraph, 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 ), ) .. _custom-media-json-encoder: .. rubric:: Custom JSON encoder You can also override the default :class:`~json.JSONEncoder` by using a custom Encoder and updating the media handlers for ``application/json`` type to use that:: import json from datetime import datetime from functools import partial import falcon from falcon import media class DatetimeEncoder(json.JSONEncoder): \"\"\"Json Encoder that supports datetime objects.\"\"\" def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return super().default(obj) app = falcon.App() json_handler = media.JSONHandler( dumps=partial(json.dumps, cls=DatetimeEncoder), ) extra_handlers = { 'application/json': json_handler, } app.req_options.media_handlers.update(extra_handlers) app.resp_options.media_handlers.update(extra_handlers) .. note:: When testing an application employing a custom JSON encoder, bear in mind that :class:`~.testing.TestClient` is decoupled from the app, and it simulates requests as if they were performed by a third-party client (just sans network). Therefore, passing the **json** parameter to :ref:`simulate_* <testing_standalone_methods>` methods will effectively use the stdlib's :func:`json.dumps`. If you want to serialize custom objects for testing, you will need to dump them into a string yourself, and pass it using the **body** parameter instead (accompanied by the ``application/json`` content type header). 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() # NOTE(kgriffs): Make content_type a kwarg to support the # Request.render_body() shortcut optimization. def _serialize_b(self, media, content_type=None) -> bytes: return self._dumps(media) async def _serialize_async_b(self, media, content_type) -> bytes: return self._dumps(media)
[文档]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