Serialization

Restate sends data over the network for storing state, journaling actions, awakeables, etc. There are multiple ways to specify which (de)serializers should be used.

Default Serialization

By default, payloads are serialized with bytes(json.dumps(obj), "utf-8") and deserialized with json.loads(buf). If this does not work for your data type, then you need to specify a custom serializer, as shown below.

Custom Serialization

To write a custom serializer, you implement the Serde interface.

For example a custom JSON serializer could look like this:

class MyData(typing.TypedDict):
    """Represents a response from the GPT model."""
    some_value: str
    my_number: int


class MySerde(Serde[MyData]):
    def deserialize(self, buf: bytes) -> typing.Optional[MyData]:
        if not buf:
            return None
        data = json.loads(buf)
        return MyData(some_value=data["some_value"], my_number=data["some_number"])

    def serialize(self, obj: typing.Optional[MyData]) -> bytes:
        if obj is None:
            return bytes()
        data = {
            "some_value": obj["some_value"],
            "some_number": obj["my_number"]
        }
        return bytes(json.dumps(data), "utf-8")

You then use this serializer in your handlers, as follows:

# For the input/output serialization of your handlers
@my_object.handler(input_serde=MySerde(), output_serde=MySerde())
async def my_handler(ctx: ObjectContext, greeting: str) -> str:

    # To serialize state
    await ctx.get("my_state", serde=MySerde())
    ctx.set("my_state", MyData(some_value="value", my_number=123), serde=MySerde())

    # To serialize awakeable payloads
    ctx.awakeable(serde=MySerde())

    # To serialize the results of actions
    await ctx.run("some-task", some_task, serde=MySerde())

    # etc.

    return "some-output"