Expand source code
class SocketModeRequest:
type: str
envelope_id: str
payload: dict
accepts_response_payload: bool
retry_attempt: Optional[int] # events_api
retry_reason: Optional[str] # events_api
def __init__(
self,
type: str,
envelope_id: str,
payload: Union[dict, JsonObject, str],
accepts_response_payload: Optional[bool] = None,
retry_attempt: Optional[int] = None,
retry_reason: Optional[str] = None,
):
self.type = type
self.envelope_id = envelope_id
if isinstance(payload, JsonObject):
self.payload = payload.to_dict()
elif isinstance(payload, dict):
self.payload = payload
elif isinstance(payload, str):
self.payload = {"text": payload}
else:
unexpected_payload_type = type(payload) # type: ignore
raise ValueError(f"Unsupported payload data type ({unexpected_payload_type})")
self.accepts_response_payload = accepts_response_payload or False
self.retry_attempt = retry_attempt
self.retry_reason = retry_reason
@classmethod
def from_dict(cls, message: dict) -> Optional["SocketModeRequest"]:
if all(k in message for k in ("type", "envelope_id", "payload")):
return SocketModeRequest(
type=message.get("type"),
envelope_id=message.get("envelope_id"),
payload=message.get("payload"),
accepts_response_payload=message.get("accepts_response_payload") or False,
retry_attempt=message.get("retry_attempt"),
retry_reason=message.get("retry_reason"),
)
return None
def to_dict(self) -> dict: # skipcq: PYL-W0221
d = {"envelope_id": self.envelope_id}
if self.payload is not None:
d["payload"] = self.payload
return d