ES shared chat sessions

Author: chillymosh

twitchio/eventsub/payloads.py

@@ -362,6 +362,51 @@ def condition(self) -> Condition:
         return {"broadcaster_user_id": self.broadcaster_user_id, "user_id": self.user_id}
 
 
+class SharedChatSessionBeginSubscription(SubscriptionPayload):
+    type: ClassVar[Literal["channel.shared_chat.begin"]] = "channel.shared_chat.begin"
+    version: ClassVar[Literal["1"]] = "1"
+
+    def __init__(self, **condition: Unpack[Condition]) -> None:
+        self.broadcaster_user_id: str = condition.get("broadcaster_user_id", "")
+
+        if not self.broadcaster_user_id:
+            raise ValueError('The parameter "broadcaster_user_id" must be passed.')
+
+    @property
+    def condition(self) -> Condition:
+        return {"broadcaster_user_id": self.broadcaster_user_id}
+
+
+class SharedChatSessionUpdateSubscription(SubscriptionPayload):
+    type: ClassVar[Literal["channel.shared_chat.update"]] = "channel.shared_chat.update"
+    version: ClassVar[Literal["1"]] = "1"
+
+    def __init__(self, **condition: Unpack[Condition]) -> None:
+        self.broadcaster_user_id: str = condition.get("broadcaster_user_id", "")
+
+        if not self.broadcaster_user_id:
+            raise ValueError('The parameter "broadcaster_user_id" must be passed.')
+
+    @property
+    def condition(self) -> Condition:
+        return {"broadcaster_user_id": self.broadcaster_user_id}
+
+
+class SharedChatSessionEndSubscription(SubscriptionPayload):
+    type: ClassVar[Literal["channel.shared_chat.end"]] = "channel.shared_chat.end"
+    version: ClassVar[Literal["1"]] = "1"
+
+    def __init__(self, **condition: Unpack[Condition]) -> None:
+        self.broadcaster_user_id: str = condition.get("broadcaster_user_id", "")
+
+        if not self.broadcaster_user_id:
+            raise ValueError('The parameter "broadcaster_user_id" must be passed.')
+
+    @property
+    def condition(self) -> Condition:
+        return {"broadcaster_user_id": self.broadcaster_user_id}
+
+
 class ChannelSubscribeSubscription(SubscriptionPayload):
     type: ClassVar[Literal["channel.subscribe"]] = "channel.subscribe"
     version: ClassVar[Literal["1"]] = "1"

twitchio/models/eventsub_.py

@@ -1496,6 +1496,114 @@ def full_message(self) -> str:
         return " ".join(fragment.text for fragment in self.fragments if fragment.type == "text")
 
 
+class BaseSharedChatSession(BaseEvent):
+    __slots__ = (
+        "session_id",
+        "broadcaster",
+        "host",
+    )
+
+    def __init__(
+        self,
+        payload: ChannelSharedChatSessionBeginEvent | ChannelSharedChatSessionUpdateEvent | ChannelSharedChatSessionEndEvent,
+        *,
+        http: HTTPClient,
+    ) -> None:
+        self.session_id: str = payload["session_id"]
+        self.broadcaster: PartialUser = PartialUser(
+            payload["broadcaster_user_id"], payload["broadcaster_user_login"], http=http
+        )
+        self.host: PartialUser = PartialUser(
+            payload["host_broadcaster_user_id"], payload["host_broadcaster_user_login"], http=http
+        )
+
+    def __repr__(self) -> str:
+        return f"<BaseSharedChatSession session_id={self.session_id} broadcaster={self.broadcaster} host={self.host}>"
+
+
+class SharedChatSessionBegin(BaseSharedChatSession):
+    """
+    Represents a shared chat session begin event.
+
+    Attributes
+    ----------
+    session_id: PartialUser
+        The unique identifier for the shared chat session.
+    broadcaster: PartialUser
+        The user of the channel in the subscription condition which is now active in the shared chat session.
+    host: PartialUser
+        The user of the host channel.
+    participants: str
+        List of participants in the session.
+    """
+
+    subscription_type = "channel.shared_chat.begin"
+
+    __slots__ = ("participants",)
+
+    def __init__(self, payload: ChannelSharedChatSessionBeginEvent, *, http: HTTPClient) -> None:
+        super().__init__(payload, http=http)
+        self.participants: list[PartialUser] = [
+            PartialUser(p["broadcaster_user_id"], p["broadcaster_user_login"], http=http) for p in payload["participants"]
+        ]
+
+    def __repr__(self) -> str:
+        return f"<BaseSharedChatSession session_id={self.session_id} broadcaster={self.broadcaster} host={self.host}>"
+
+
+class SharedChatSessionUpdate(BaseSharedChatSession):
+    """
+    Represents a shared chat session begin event.
+
+    Attributes
+    ----------
+    session_id: PartialUser
+        The unique identifier for the shared chat session.
+    broadcaster: PartialUser
+        The user of the channel in the subscription condition which is now active in the shared chat session.
+    host: PartialUser
+        The user of the host channel.
+    participants: str
+        List of participants in the session.
+    """
+
+    subscription_type = "channel.shared_chat.update"
+
+    __slots__ = ("participants",)
+
+    def __init__(self, payload: ChannelSharedChatSessionUpdateEvent, *, http: HTTPClient) -> None:
+        super().__init__(payload, http=http)
+        self.participants: list[PartialUser] = [
+            PartialUser(p["broadcaster_user_id"], p["broadcaster_user_login"], http=http) for p in payload["participants"]
+        ]
+
+    def __repr__(self) -> str:
+        return f"<SharedChatSessionUpdate session_id={self.session_id} broadcaster={self.broadcaster} host={self.host}>"
+
+
+class SharedChatSessionEnd(BaseSharedChatSession):
+    """
+    Represents a shared chat session end event.
+
+    Attributes
+    ----------
+    session_id: PartialUser
+        The unique identifier for the shared chat session.
+    broadcaster: PartialUser
+        The user of the channel in the subscription condition which is no longer active in the shared chat session.
+    host: PartialUser
+        The user of the host channel.
+    """
+
+    subscription_type = "channel.shared_chat.end"
+
+    def __init__(self, payload: ChannelSharedChatSessionUpdateEvent, *, http: HTTPClient) -> None:
+        super().__init__(payload, http=http)
+
+    def __repr__(self) -> str:
+        return f"<SharedChatSessionEnd session_id={self.session_id} broadcaster={self.broadcaster} host={self.host}>"
+
+
 class ChannelSubscribe(BaseEvent):
     """
     Represents a channel subscribe event.

twitchio/types_/eventsub.py

@@ -126,6 +126,9 @@
     "HypeTrainProgressEvent",
     "HypeTrainEndEvent",
     "HypeTrainContributionData",
+    "ChannelSharedChatSessionBeginEvent",
+    "ChannelSharedChatSessionUpdateEvent",
+    "ChannelSharedChatSessionEndEvent",
     "ShieldModeBeginEvent",
     "ShieldModeEndEvent",
     "ShoutoutCreateEvent",
@@ -566,6 +569,24 @@ class ChatUserMessageUpdateEvent(ChatUserMessageHoldEvent):
     status: Literal["approved", "denied", "invalid"]
 
 
+class BaseSharedChatSessionData(BaseBroadcasterEvent):
+    session_id: str
+    host_broadcaster_user_id: str
+    host_broadcaster_user_login: str
+    host_broadcaster_user_name: str
+
+
+class ChannelSharedChatSessionBeginEvent(BaseSharedChatSessionData):
+    participants: list[BaseBroadcasterEvent]
+
+
+class ChannelSharedChatSessionUpdateEvent(BaseSharedChatSessionData):
+    participants: list[BaseBroadcasterEvent]
+
+
+class ChannelSharedChatSessionEndEvent(BaseSharedChatSessionData): ...
+
+
 class ChannelSubscribeEvent(BroadcasterUserEvent):
     tier: Literal["1000", "2000", "3000"]
     is_gift: bool