Skip to content

declearn.messaging.Message

Abstract base dataclass to define parsable messages.

A 'Message' is merely an arbitrary data structure that implements conversion to and from a JSON-serializable dict, and is associated with a unique typekey string class attribute under which it is type-registered.

All subclasses must be decorated into a dataclasses.dataclass.

Subclasses that only have serializable fields do not need to define anything else than typekey. If type-conversion is required, they may overload the to_kwargs method and from_kwargs classmethod.

Subclasses are type-registered by default. This can be prevented (e.g. in testing contexts, or when defining an abstract subclass) by passing the register=False keyword argument at inheritance; e.g. class MyMsg(Message, register=False):.

Source code in declearn/messaging/_api.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@create_types_registry(name="Message")
@dataclasses.dataclass
class Message(metaclass=ABCMeta):
    """Abstract base dataclass to define parsable messages.

    A 'Message' is merely an arbitrary data structure that implements
    conversion to and from a JSON-serializable dict, and is associated
    with a unique `typekey` string class attribute under which it is
    type-registered.

    All subclasses must be decorated into a `dataclasses.dataclass`.

    Subclasses that only have serializable fields do not need to define
    anything else than `typekey`. If type-conversion is required, they
    may overload the `to_kwargs` method and `from_kwargs` classmethod.

    Subclasses are type-registered by default. This can be prevented
    (e.g. in testing contexts, or when defining an abstract subclass)
    by passing the `register=False` keyword argument at inheritance;
    e.g. `class MyMsg(Message, register=False):`.
    """

    typekey: ClassVar[str]

    def __init_subclass__(
        cls,
        register: bool = True,
    ) -> None:
        """Automatically type-register subclasses."""
        if register:
            register_type(cls, name=cls.typekey, group="Message")

    def to_kwargs(self) -> Dict[str, Any]:
        """Return a JSON-serializable dict representation of this message."""
        # NOTE: override this method to serialize attributes
        #       that are not handled by declearn.utils.json_unpack
        return dataclasses.asdict(self)

    @classmethod
    def from_kwargs(cls, **kwargs: Any) -> Self:
        """Parse the message from JSON-deserialized attributes."""
        # NOTE: override this method to de-serialize attributes
        #       that are not handled by declearn.utils.json_pack
        return cls(**kwargs)

    def to_string(self) -> str:
        """Convert the message to a JSON-serialized string."""
        data = self.to_kwargs()
        dump = json.dumps(data, default=json_pack)
        return self.typekey + "\n" + dump

__init_subclass__(register=True)

Automatically type-register subclasses.

Source code in declearn/messaging/_api.py 
65
66
67
68
69
70
71
def __init_subclass__(
    cls,
    register: bool = True,
) -> None:
    """Automatically type-register subclasses."""
    if register:
        register_type(cls, name=cls.typekey, group="Message")

from_kwargs(**kwargs) classmethod

Parse the message from JSON-deserialized attributes.

Source code in declearn/messaging/_api.py 
79
80
81
82
83
84
@classmethod
def from_kwargs(cls, **kwargs: Any) -> Self:
    """Parse the message from JSON-deserialized attributes."""
    # NOTE: override this method to de-serialize attributes
    #       that are not handled by declearn.utils.json_pack
    return cls(**kwargs)

to_kwargs()

Return a JSON-serializable dict representation of this message.

Source code in declearn/messaging/_api.py 
73
74
75
76
77
def to_kwargs(self) -> Dict[str, Any]:
    """Return a JSON-serializable dict representation of this message."""
    # NOTE: override this method to serialize attributes
    #       that are not handled by declearn.utils.json_unpack
    return dataclasses.asdict(self)

to_string()

Convert the message to a JSON-serialized string.

Source code in declearn/messaging/_api.py 
86
87
88
89
90
def to_string(self) -> str:
    """Convert the message to a JSON-serialized string."""
    data = self.to_kwargs()
    dump = json.dumps(data, default=json_pack)
    return self.typekey + "\n" + dump