Serializers

from flexiq import Queue, JsonSerializer, CborSerializer

The Rust core treats payloads as opaque bytes — serialization happens entirely Python-side. Producers and workers must use a compatible serializer.

Built-ins

Set one queue-wide, or override per task:

queue = Queue(serializer=JsonSerializer())

@queue.task(serializer=CborSerializer())
def cross_sdk_job(payload: dict) -> None: ...

Signing and encryption

SignedSerializer(inner, key) and EncryptedSerializer(inner, key) both wrap a delegate serializer. key must be at least 32 bytes of CSPRNG output, shared by every producer and worker. Signing authenticates (prevents a storage writer from smuggling code into a cloudpickle payload); encryption also gives confidentiality.

import os
from flexiq import Queue, SignedSerializer, SmartSerializer

key = os.urandom(32)  # share across producers and workers
queue = Queue(serializer=SignedSerializer(SmartSerializer(), key))

Payload codecs

A PayloadCodec is a reversible byte transform layered around a serializer — compression, encryption, signing — rather than a replacement for it. Codecs compose: a chain encodes in list order on the producer and decodes in reverse on the worker. Wire formats are part of the cross-SDK contract, so a codec-framed payload decodes from any FlexiQ SDK.

Apply a chain queue-wide with Queue(codec=...) (wraps the serializer, covers every payload and result), or register named codecs via Queue(codecs=...) and opt individual tasks in with @queue.task(codecs=[...]) (payload only — results still use the plain serializer):

from flexiq import Queue, GzipCodec, HmacCodec

queue = Queue(codec=[GzipCodec(), HmacCodec(hmac_key)])  # gzip, then hmac

Serializer protocol

class Serializer(Protocol):
    def dumps(self, obj: Any) -> bytes: ...
    def loads(self, data: bytes) -> Any: ...

Any object with dumps/loads is a valid serializer. See the Serializers guide for a custom implementation and the format-selection tradeoffs.