Source code for falcon.media.msgpack
from __future__ import annotations
from typing import Any, Callable, Optional, Protocol
from falcon import errors
from falcon.media.base import BaseHandler
from falcon.media.base import BinaryBaseHandlerWS
from falcon.typing import AsyncReadableIO
from falcon.typing import ReadableIO
[docs]
class MessagePackHandler(BaseHandler):
"""Handler built using the :mod:`msgpack` module.
This handler uses ``msgpack.unpackb()`` and ``msgpack.Packer().pack()``. The
MessagePack ``bin`` type is used to distinguish between Unicode strings
(of type ``str``) and byte strings (of type ``bytes``).
This handler will raise a :class:`falcon.MediaNotFoundError` when attempting
to parse an empty body; it will raise a :class:`falcon.MediaMalformedError`
if an error happens while parsing the body.
Note:
This handler requires the extra ``msgpack`` package (version 0.5.2
or higher), which must be installed in addition to ``falcon`` from
PyPI:
.. code::
$ pip install msgpack
"""
_pack: Callable[[Any], bytes]
_unpackb: UnpackMethod
def __init__(self) -> None:
import msgpack
packer = msgpack.Packer(autoreset=True, use_bin_type=True)
self._pack = packer.pack
self._unpackb = msgpack.unpackb
# NOTE(kgriffs): To be safe, only enable the optimized protocol when
# not subclassed.
if type(self) is MessagePackHandler:
self._serialize_sync = self._pack # type: ignore[assignment]
self._deserialize_sync = self._deserialize
def _deserialize(self, data: bytes) -> Any:
if not data:
raise errors.MediaNotFoundError('MessagePack')
try:
# NOTE(jmvrbanac): Using unpackb since we would need to manage
# a buffer for Unpacker() which wouldn't gain us much.
return self._unpackb(data, raw=False)
except ValueError as err:
raise errors.MediaMalformedError('MessagePack') from err
def deserialize(
self,
stream: ReadableIO,
content_type: Optional[str],
content_length: Optional[int],
) -> Any:
return self._deserialize(stream.read())
async def deserialize_async(
self,
stream: AsyncReadableIO,
content_type: Optional[str],
content_length: Optional[int],
) -> Any:
return self._deserialize(await stream.read())
def serialize(self, media: Any, content_type: Optional[str]) -> bytes:
return self._pack(media)
async def serialize_async(self, media: Any, content_type: Optional[str]) -> bytes:
return self._pack(media)
[docs]
class MessagePackHandlerWS(BinaryBaseHandlerWS):
"""WebSocket media handler for de(serializing) MessagePack to/from BINARY payloads.
This handler uses ``msgpack.unpackb()`` and ``msgpack.packb()``. The
MessagePack ``bin`` type is used to distinguish between Unicode strings
(of type ``str``) and byte strings (of type ``bytes``).
Note:
This handler requires the extra ``msgpack`` package (version 0.5.2
or higher), which must be installed in addition to ``falcon`` from
PyPI:
.. code::
$ pip install msgpack
"""
__slots__ = ('msgpack', 'packer')
_pack: Callable[[Any], bytes]
_unpackb: UnpackMethod
def __init__(self) -> None:
import msgpack
packer = msgpack.Packer(autoreset=True, use_bin_type=True)
self._pack = packer.pack
self._unpackb = msgpack.unpackb
def serialize(self, media: object) -> bytes:
return self._pack(media)
def deserialize(self, payload: bytes) -> Any:
# NOTE(jmvrbanac): Using unpackb since we would need to manage
# a buffer for Unpacker() which wouldn't gain us much.
return self._unpackb(payload, raw=False)
class UnpackMethod(Protocol):
def __call__(self, data: bytes, raw: bool = ...) -> Any: ...