Add Shield events Eventsub

Add Shield events Eventsub

Author: chillymosh

docs/changelog.rst

@@ -24,6 +24,9 @@ Master
 - ext.eventsub
     - Additions
         - Updated docs regarding new HypeTrain contribution method ``other`` for :attr:`~twitchio.ext.eventsub.HypeTrainContributor.type`
+        - Added Shield Status events
+            - :func:`~twitchio.ext.eventsub.EventSubClient.subscribe_channel_shield_mode_begin`
+            - :func:`~twitchio.ext.eventsub.EventSubClient.subscribe_channel_shield_mode_end`
 
 2.5.0
 ======

docs/exts/eventsub.rst

@@ -192,6 +192,14 @@ This is a list of events dispatched by the eventsub ext.
 
     Called when a hype train ends on their channel.
 
+.. function:: event_eventsub_notification_channel_shield_mode_begin(event: ShieldModeBegin)
+
+    Called when a channel's Shield Mode status is activated.
+
+.. function:: event_eventsub_notification_channel_shield_mode_end(event: ShieldModeEnd)
+
+    Called when a channel's Shield Mode status is deactivated.
+
 API Reference
 --------------
 
@@ -387,6 +395,18 @@ API Reference
     :members:
     :inherited-members:
 
+.. attributetable::: ShieldModeBegin
+
+.. autoclass:: ShieldModeBegin
+    :members:
+    :inherited-members:
+
+.. attributetable::: ShieldModeEnd
+
+.. autoclass:: ShieldModeEnd
+    :members:
+    :inherited-members:
+
 .. attributetable::: StreamOnlineData
 
 .. autoclass:: StreamOnlineData

twitchio/ext/eventsub/models.py

@@ -1314,6 +1314,50 @@ def __init__(self, client: EventSubClient, data: dict):
         self.ended_at: datetime.datetime = _parse_datetime(data["ended_at"])
 
 
+class ShieldModeBegin:
+    """
+    Represents a Shield Mode activation status.
+
+    Attributes
+    -----------
+    broadcaster: :class:`~twitchio.PartialUser`
+        The broadcaster whose Shield Mode status was updated.
+    moderator: :class:`~twitchio.PartialUser`
+        The moderator that updated the Shield Mode staus.
+    started_at: :class:`datetime.datetime`
+        The UTC datetime of when Shield Mode was last activated.
+    """
+
+    __slots__ = ("broadcaster", "moderator", "started_at")
+
+    def __init__(self, client: EventSubClient, data: dict):
+        self.broadcaster: PartialUser = _transform_user(client, data, "broadcaster_user")
+        self.moderator: PartialUser = _transform_user(client, data, "moderator_user")
+        self.started_at: datetime.datetime = _parse_datetime(data["started_at"])
+
+
+class ShieldModeEnd:
+    """
+    Represents a Shield Mode activation status.
+
+    Attributes
+    -----------
+    broadcaster: :class:`~twitchio.PartialUser`
+        The broadcaster whose Shield Mode status was updated.
+    moderator: :class:`~twitchio.PartialUser`
+        The moderator that updated the Shield Mode staus.
+    ended_at: :class:`datetime.datetime`
+        The UTC datetime of when Shield Mode was last deactivated.
+    """
+
+    __slots__ = ("broadcaster", "moderator", "ended_at")
+
+    def __init__(self, client: EventSubClient, data: dict):
+        self.broadcaster: PartialUser = _transform_user(client, data, "broadcaster_user")
+        self.moderator: PartialUser = _transform_user(client, data, "moderator_user")
+        self.ended_at: datetime.datetime = _parse_datetime(data["ended_at"])
+
+
 _DataType = Union[
     ChannelBanData,
     ChannelUnbanData,
@@ -1342,6 +1386,8 @@ def __init__(self, client: EventSubClient, data: dict):
     UserAuthorizationGrantedData,
     UserAuthorizationRevokedData,
     UserUpdateData,
+    ShieldModeBegin,
+    ShieldModeEnd,
 ]
 
 
@@ -1387,6 +1433,9 @@ class _SubscriptionTypes(metaclass=_SubTypesMeta):
     channel_goal_progress = "channel.goal.progress", 1, ChannelGoalBeginProgressData
     channel_goal_end = "channel.goal.end", 1, ChannelGoalEndData
 
+    channel_shield_mode_begin = "channel.shield_mode.begin", 1, ShieldModeBegin
+    channel_shield_mode_end = "channel.shield_mode.end", 1, ShieldModeEnd
+
     hypetrain_begin = "channel.hype_train.begin", 1, HypeTrainBeginProgressData
     hypetrain_progress = "channel.hype_train.progress", 1, HypeTrainBeginProgressData
     hypetrain_end = "channel.hype_train.end", 1, HypeTrainEndData

twitchio/ext/eventsub/server.py

@@ -111,6 +111,23 @@ async def _subscribe_with_broadcaster(
         broadcaster = str(broadcaster)
         return await self._http.create_subscription(event, {"broadcaster_user_id": broadcaster})
 
+    async def _subscribe_with_broadcaster_moderator(
+        self,
+        event: Tuple[str, int, Type[models._DataType]],
+        broadcaster: Union[PartialUser, str, int],
+        moderator: Union[PartialUser, str, int],
+    ):
+        if isinstance(broadcaster, PartialUser):
+            broadcaster = broadcaster.id
+        if isinstance(moderator, PartialUser):
+            moderator = moderator.id
+
+        broadcaster = str(broadcaster)
+        moderator = str(moderator)
+        return await self._http.create_subscription(
+            event, {"broadcaster_user_id": broadcaster, "moderator_user_id": moderator}
+        )
+
     def subscribe_channel_bans(self, broadcaster: Union[PartialUser, str, int]):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.ban, broadcaster)
 
@@ -214,6 +231,20 @@ def subscribe_channel_prediction_lock(self, broadcaster: Union[PartialUser, str,
     def subscribe_channel_prediction_end(self, broadcaster: Union[PartialUser, str, int]):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.prediction_end, broadcaster)
 
+    def subscribe_channel_shield_mode_begin(
+        self, broadcaster: Union[PartialUser, str, int], moderator: Union[PartialUser, str, int]
+    ):
+        return self._subscribe_with_broadcaster_moderator(
+            models.SubscriptionTypes.channel_shield_mode_begin, broadcaster, moderator
+        )
+
+    def subscribe_channel_shield_mode_end(
+        self, broadcaster: Union[PartialUser, str, int], moderator: Union[PartialUser, str, int]
+    ):
+        return self._subscribe_with_broadcaster_moderator(
+            models.SubscriptionTypes.channel_shield_mode_end, broadcaster, moderator
+        )
+
     async def subscribe_user_authorization_granted(self):
         return await self._http.create_subscription(
             models.SubscriptionTypes.user_authorization_grant, {"client_id": self.client._http.client_id}