Handling payload data

The response.payload module provides a basic interface and implementations for handling payload data returned by a HTTP request.

Basic usage

The PayloadHandler transforms data returned by a HTTP request to a usable python object.

import asyncio
from typing import Any

from aiorequestful.response.payload import PayloadHandler


async def handle(handler: PayloadHandler, payload: Any) -> None:
    print(await handler.serialize(payload))  # convert the payload data to a string
    print(await handler.deserialize(payload))  # convert the payload data to the required object type

asyncio.run(handle(handler=PayloadHandler(), payload='{"key": "value"}'))

These two methods should accept a variety of input types as below where T is the supported output type of the PayloadHandler. Crucially, PayloadHandler.deserialize() should accept the ClientResponse as returned by the aiohttp package.

@abstractmethod
async def serialize(self, payload: str | bytes | bytearray | T) -> str:
    """Serialize the payload object to a string."""
    raise NotImplementedError

@abstractmethod
async def deserialize(self, response: str | bytes | bytearray | ClientResponse | T) -> T:
    """
    Extract payload data from the given ``response`` and serialize to the appropriate object.

    :param response: The response/payload to handle.
    :raise PayloadHandlerError: When the input data is not recognised.
    """
    raise NotImplementedError

See also

This module implements a few common payload data types as shown below, though you may wish to extend this functionality.

See also

For more info on how to pass a PayloadHandler to the RequestHandler, see Handling the response payload.

StringPayloadHandler

Converts payload data to str objects.

from aiorequestful.response.payload import StringPayloadHandler

payload_data = {"key": "value"}
payload_handler = StringPayloadHandler()


async def handle(handler: PayloadHandler, payload: Any) -> None:
    print(await handler.serialize(payload))  # convert the payload data to a string
    print(await handler.deserialize(payload))  # convert the payload data to a string

asyncio.run(handle(handler=payload_handler, payload=payload_data))

JSONPayloadHandler

Converts payload data to dict objects.

from aiorequestful.response.payload import JSONPayloadHandler

payload_data = '{"key": "value"}'
payload_handler = JSONPayloadHandler()


async def handle(handler: PayloadHandler, payload: Any) -> None:
    print(await handler.serialize(payload))  # convert the payload data to a string
    print(await handler.deserialize(payload))  # convert the payload data to a dict

asyncio.run(handle(handler=payload_handler, payload=payload_data))

Writing a PayloadHandler

To implement a PayloadHandler, you will need to implement the abstract methods as shown below.

class PayloadHandler[T: Any](ABC):
    """Handles payload data conversion to return response payload in expected format."""

    __slots__ = ()

    @abstractmethod
    async def serialize(self, payload: str | bytes | bytearray | T) -> str:
        """Serialize the payload object to a string."""
        raise NotImplementedError

    @abstractmethod
    async def deserialize(self, response: str | bytes | bytearray | ClientResponse | T) -> T:
        """
        Extract payload data from the given ``response`` and serialize to the appropriate object.

        :param response: The response/payload to handle.
        :raise PayloadHandlerError: When the input data is not recognised.
        """
        raise NotImplementedError

    def __call__(self, response: str | bytes | bytearray | ClientResponse | T) -> Awaitable[T]:
        return self.deserialize(response=response)

As an example, the following implements the JSONPayloadHandler.

class JSONPayloadHandler(PayloadHandler[JSON]):

    __slots__ = ("indent",)

    def __init__(self, indent: int = None):
        self.indent = indent

    async def serialize(self, payload: str | bytes | bytearray | JSON) -> str:
        if isinstance(payload, str | bytes | bytearray):
            try:
                payload = json.loads(payload)
            except (json.decoder.JSONDecodeError, TypeError):
                raise PayloadHandlerError(f"Unrecognised input type: {payload}")
        return json.dumps(payload, indent=self.indent)

    async def deserialize(self, response: str | bytes | bytearray | ClientResponse | JSON) -> JSON:
        match response:
            case dict():
                try:  # check the given payload can be converted to/from JSON format
                    return json.loads(json.dumps(response))
                except (json.decoder.JSONDecodeError, TypeError):
                    raise PayloadHandlerError("Given payload is not a valid JSON object")
            case str() | bytes() | bytearray():
                return json.loads(response)
            case ClientResponse():
                return await response.json(content_type=None)
            case _:
                raise PayloadHandlerError(f"Unrecognised input type: {response}")