"""Represents a Handshake Reuse message under RFC 0434."""
from typing import Optional, Text
from marshmallow import EXCLUDE, ValidationError, fields, pre_dump
from .....messaging.agent_message import AgentMessage, AgentMessageSchema
from ..message_types import DEFAULT_VERSION, MESSAGE_REUSE, PROTOCOL_PACKAGE
HANDLER_CLASS = (
f"{PROTOCOL_PACKAGE}.handlers.reuse_handler.HandshakeReuseMessageHandler"
)
[docs]class HandshakeReuse(AgentMessage):
"""Class representing a Handshake Reuse message."""
def __init__(
self,
version: str = DEFAULT_VERSION,
msg_type: Optional[Text] = None,
**kwargs,
):
"""Initialize Handshake Reuse message object."""
super().__init__(_type=msg_type, _version=version, **kwargs)
[docs]class HandshakeReuseSchema(AgentMessageSchema):
"""Handshake Reuse schema class."""
_type = fields.Str(
data_key="@type",
required=False,
metadata={
"description": "Message type",
"example": "https://didcomm.org/my-family/1.0/my-message-type",
},
)
@pre_dump
def check_thread_deco(self, obj, **kwargs):
"""Thread decorator, and its thid and pthid, are mandatory."""
if not obj._decorators.to_dict().get("~thread", {}).keys() >= {"thid", "pthid"}:
raise ValidationError("Missing required field(s) in thread decorator")
return obj