import abc
import io
from typing import Union
from falcon.constants import MEDIA_JSON
[docs]
class BaseHandler(metaclass=abc.ABCMeta):
"""Abstract Base Class for an internet media type handler."""
# NOTE(kgriffs): The following special methods are used to enable an
# optimized media (de)serialization protocol for ASGI. This is not
# currently part of the public interface for Falcon, and is not
# included in the docs. Once we are happy with the protocol, we
# might make it part of the public interface for use by custom
# media type handlers.
_serialize_sync = None
"""Override to provide a synchronous serialization method that takes an object."""
_deserialize_sync = None
"""Override to provide a synchronous deserialization method that
takes a byte string."""
[docs]
def serialize(self, media, content_type) -> bytes:
"""Serialize the media object on a :any:`falcon.Response`.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden in order to work with WSGI apps. Child classes
can ignore this method if they are only to be used
with ASGI apps, as long as they override
:py:meth:`~.BaseHandler.serialize_async`.
Note:
The JSON media handler is an exception in requiring the implementation of
the sync version also for ASGI apps. See the
:ref:`this section<note_json_handler>` for more details.
Args:
media (object): A serializable object.
content_type (str): Type of response content.
Returns:
bytes: The resulting serialized bytes from the input object.
"""
if MEDIA_JSON in content_type:
raise NotImplementedError(
'The JSON media handler requires the sync interface to be '
"implemented even in ASGI applications, because it's used "
'internally by the Falcon framework.'
)
else:
raise NotImplementedError()
[docs]
async def serialize_async(self, media, content_type) -> bytes:
"""Serialize the media object on a :any:`falcon.Response`.
This method is similar to :py:meth:`~.BaseHandler.serialize`
except that it is asynchronous. The default implementation simply calls
:py:meth:`~.BaseHandler.serialize`. If the media object may be
awaitable, or is otherwise something that should be read
asynchronously, subclasses must override the default implementation
in order to handle that case.
Note:
By default, the :py:meth:`~.BaseHandler.serialize`
method raises an instance of :py:class:`NotImplementedError`.
Therefore, child classes must either override
:py:meth:`~.BaseHandler.serialize` or
:py:meth:`~.BaseHandler.serialize_async` in order to be
compatible with ASGI apps.
Args:
media (object): A serializable object.
content_type (str): Type of response content.
Returns:
bytes: The resulting serialized bytes from the input object.
"""
return self.serialize(media, content_type)
[docs]
def deserialize(self, stream, content_type, content_length) -> object:
"""Deserialize the :any:`falcon.Request` body.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden in order to work with WSGI apps. Child classes
can ignore this method if they are only to be used
with ASGI apps, as long as they override
:py:meth:`~.BaseHandler.deserialize_async`.
Note:
The JSON media handler is an exception in requiring the implementation of
the sync version also for ASGI apps. See the
:ref:`this section<note_json_handler>` for more details.
Args:
stream (object): Readable file-like object to deserialize.
content_type (str): Type of request content.
content_length (int): Length of request content.
Returns:
object: A deserialized object.
"""
if MEDIA_JSON in content_type:
raise NotImplementedError(
'The JSON media handler requires the sync interface to be '
"implemented even in ASGI applications, because it's used "
'internally by the Falcon framework.'
)
else:
raise NotImplementedError()
[docs]
async def deserialize_async(self, stream, content_type, content_length) -> object:
"""Deserialize the :any:`falcon.Request` body.
This method is similar to :py:meth:`~.BaseHandler.deserialize` except
that it is asynchronous. The default implementation adapts the
synchronous :py:meth:`~.BaseHandler.deserialize` method
via :py:class:`io.BytesIO`. For improved performance, media handlers should
override this method.
Note:
By default, the :py:meth:`~.BaseHandler.deserialize`
method raises an instance of :py:class:`NotImplementedError`.
Therefore, child classes must either override
:py:meth:`~.BaseHandler.deserialize` or
:py:meth:`~.BaseHandler.deserialize_async` in order to be
compatible with ASGI apps.
Args:
stream (object): Asynchronous file-like object to deserialize.
content_type (str): Type of request content.
content_length (int): Length of request content.
Returns:
object: A deserialized object.
"""
data = await stream.read()
# NOTE(kgriffs): Override content length to make sure it is correct,
# since we know what it is in this case.
content_length = len(data)
return self.deserialize(io.BytesIO(data), content_type, content_length)
exhaust_stream = False
"""Whether to exhaust the input stream upon finishing deserialization.
Exhausting the stream may be useful for handlers that do not necessarily
consume the whole stream, but the deserialized media object is complete and
does not involve further streaming.
"""
[docs]
class TextBaseHandlerWS(metaclass=abc.ABCMeta):
"""Abstract Base Class for a WebSocket TEXT media handler."""
[docs]
def serialize(self, media: object) -> str:
"""Serialize the media object to a Unicode string.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden if the child class wishes to support
serialization to TEXT (0x01) message payloads.
Args:
media (object): A serializable object.
Returns:
str: The resulting serialized string from the input object.
"""
raise NotImplementedError()
[docs]
def deserialize(self, payload: str) -> object:
"""Deserialize TEXT payloads from a Unicode string.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden if the child class wishes to support
deserialization from TEXT (0x01) message payloads.
Args:
payload (str): Message payload to deserialize.
Returns:
object: A deserialized object.
"""
raise NotImplementedError()
[docs]
class BinaryBaseHandlerWS(metaclass=abc.ABCMeta):
"""Abstract Base Class for a WebSocket BINARY media handler."""
[docs]
def serialize(self, media: object) -> Union[bytes, bytearray, memoryview]:
"""Serialize the media object to a byte string.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden if the child class wishes to support
serialization to BINARY (0x02) message payloads.
Args:
media (object): A serializable object.
Returns:
bytes: The resulting serialized byte string from the input
object. May be an instance of :class:`bytes`,
:class:`bytearray`, or :class:`memoryview`.
"""
raise NotImplementedError()
[docs]
def deserialize(self, payload: bytes) -> object:
"""Deserialize BINARY payloads from a byte string.
By default, this method raises an instance of
:py:class:`NotImplementedError`. Therefore, it must be
overridden if the child class wishes to support
deserialization from BINARY (0x02) message payloads.
Args:
payload (bytes): Message payload to deserialize.
Returns:
object: A deserialized object.
"""
raise NotImplementedError()
