beanis.odm.custom_encoders

Custom encoder/decoder registration system for Beanis

Allows users to register custom serialization logic for any Python type.

beanis.odm.custom_encoders.EncoderFunc

Converts object to string for Redis

beanis.odm.custom_encoders.DecoderFunc

Converts string from Redis to object

CustomEncoderRegistry

class CustomEncoderRegistry()

Global registry for custom type encoders/decoders

Example usage: from beanis import register_encoder import numpy as np import base64 import pickle

@register_encoder(np.ndarray)
def encode_numpy(arr: np.ndarray) -> str:
    return base64.b64encode(pickle.dumps(arr)).decode('utf-8')

@register_decoder(np.ndarray)
def decode_numpy(data: str) -> np.ndarray:
    return pickle.loads(base64.b64decode(data))

CustomEncoderRegistry.register_encoder

@classmethod
def register_encoder(cls, type_: Type, encoder: EncoderFunc) -> None

Register an encoder for a specific type

Arguments:

  • type_ - The Python type to encode
  • encoder - Function that converts the type to a string

CustomEncoderRegistry.register_decoder

@classmethod
def register_decoder(cls, type_: Type, decoder: DecoderFunc) -> None

Register a decoder for a specific type

Arguments:

  • type_ - The Python type to decode
  • decoder - Function that converts a string to the type

CustomEncoderRegistry.register_pair

@classmethod
def register_pair(cls, type_: Type, encoder: EncoderFunc,
                  decoder: DecoderFunc) -> None

Register both encoder and decoder for a type

Arguments:

  • type_ - The Python type
  • encoder - Function that converts the type to a string
  • decoder - Function that converts a string to the type

CustomEncoderRegistry.get_encoder

@classmethod
def get_encoder(cls, type_: Type) -> Optional[EncoderFunc]

Get encoder for a type, or None if not registered

CustomEncoderRegistry.get_decoder

@classmethod
def get_decoder(cls, type_: Type) -> Optional[DecoderFunc]

Get decoder for a type, or None if not registered

CustomEncoderRegistry.clear

@classmethod
def clear(cls) -> None

Clear all registered encoders/decoders (mainly for testing)

register_encoder

def register_encoder(type_: Type) -> Callable[[EncoderFunc], EncoderFunc]

Decorator to register a custom encoder for a type

Example:

@register_encoder(np.ndarray) def encode_numpy(arr: np.ndarray) -> str: return base64.b64encode(pickle.dumps(arr)).decode('utf-8')

register_decoder

def register_decoder(type_: Type) -> Callable[[DecoderFunc], DecoderFunc]

Decorator to register a custom decoder for a type

Example:

@register_decoder(np.ndarray) def decode_numpy(data: str) -> np.ndarray: return pickle.loads(base64.b64decode(data))

register_type

def register_type(type_: Type, encoder: EncoderFunc,
                  decoder: DecoderFunc) -> None

Register both encoder and decoder for a type (non-decorator version)

Example:

register_type( np.ndarray, encoder=lambda arr: base64.b64encode(pickle.dumps(arr)).decode(), decoder=lambda s: pickle.loads(base64.b64decode(s)) )