run black

Author: IAmTomahawkx

twitchio/ext/eventsub/http.py

@@ -18,7 +18,9 @@ def __init__(self, client: Union[EventSubClient, EventSubWSClient], token: Optio
         self._http = client.client._http
         self._token = token
 
-    async def create_webhook_subscription(self, event_type: Tuple[str, int, Type[EventData]], condition: Dict[str, str]):
+    async def create_webhook_subscription(
+        self, event_type: Tuple[str, int, Type[EventData]], condition: Dict[str, str]
+    ):
         payload = {
             "type": event_type[0],
             "version": str(event_type[1]),
@@ -28,7 +30,9 @@ async def create_webhook_subscription(self, event_type: Tuple[str, int, Type[Eve
         route = Route("POST", "eventsub/subscriptions", body=payload, token=self._token)
         return await self._http.request(route, paginate=False, force_app_token=True)
 
-    async def create_websocket_subscription(self, event_type: Tuple[str, int, Type[EventData]], condition: Dict[str, str], session_id: str, token: str):
+    async def create_websocket_subscription(
+        self, event_type: Tuple[str, int, Type[EventData]], condition: Dict[str, str], session_id: str, token: str
+    ):
         payload = {
             "type": event_type[0],
             "version": str(event_type[1]),

twitchio/ext/eventsub/models.py

@@ -54,7 +54,7 @@ def __init__(self, data: dict):
         if self.transport_method is TransportType.webhook:
             self.transport.callback: str = data["transport"]["callback"]  # type: ignore
         else:
-            self.transport.callback: str = "" # type: ignore # compatibility
+            self.transport.callback: str = ""  # type: ignore # compatibility
             self.transport.session_id: str = data["transport"]["session_id"]  # type: ignore
 
 
@@ -114,12 +114,15 @@ class WebsocketHeaders:
     timestamp: :class:`datetime.datetime`
         The timestamp the message was sent at
     """
+
     def __init__(self, frame: dict):
         meta = frame["metadata"]
         self.message_id: str = meta["message_id"]
         self.timestamp = _parse_datetime(meta["message_timestamp"])
-        self.message_type: Literal["notification", "revocation", "reconnect", "session_keepalive"] = meta["message_type"]
-        self.message_retry: int = 0 # don't make breaking changes with the Header class
+        self.message_type: Literal["notification", "revocation", "reconnect", "session_keepalive"] = meta[
+            "message_type"
+        ]
+        self.message_retry: int = 0  # don't make breaking changes with the Header class
         self.signature: str = ""
         self.subscription_type: Optional[str]
         self.subscription_version: Optional[str]
@@ -153,7 +156,9 @@ def __init__(self, client: EventSubClient, _data: str, request: web.Request):
     def __init__(self, client: EventSubWSClient, _data: dict, request: None):
         ...
 
-    def __init__(self, client: Union[EventSubClient, EventSubWSClient], _data: Union[str, dict], request: Optional[web.Request]):
+    def __init__(
+        self, client: Union[EventSubClient, EventSubWSClient], _data: Union[str, dict], request: Optional[web.Request]
+    ):
         self._client = client
         self._raw_data = _data
 
@@ -178,15 +183,14 @@ def __init__(self, client: Union[EventSubClient, EventSubWSClient], _data: Union
                 self.subscription = None
             self.setup(data["payload"])
 
-
     def setup(self, data: dict):
         pass
 
     def verify(self):
         """
         Only used in webhook transport types. Verifies the message is valid
         """
-        hmac_message = (self.headers.message_id + self.headers._raw_timestamp + self._raw_data).encode("utf-8") # type: ignore
+        hmac_message = (self.headers.message_id + self.headers._raw_timestamp + self._raw_data).encode("utf-8")  # type: ignore
         secret = self._client.secret.encode("utf-8")
         digest = hmac.new(secret, msg=hmac_message, digestmod=hashlib.sha256).hexdigest()
 
@@ -220,7 +224,7 @@ def setup(self, data: dict):
         self.challenge: str = data["challenge"]
 
     def verify(self):
-        hmac_message = (self.headers.message_id + self.headers._raw_timestamp + self._raw_data).encode("utf-8") # type: ignore
+        hmac_message = (self.headers.message_id + self.headers._raw_timestamp + self._raw_data).encode("utf-8")  # type: ignore
         secret = self._client.secret.encode("utf-8")
         digest = hmac.new(secret, msg=hmac_message, digestmod=hashlib.sha256).hexdigest()
 
@@ -261,6 +265,7 @@ class KeepAliveEvent(BaseEvent):
         These are only dispatched when using :class:`~twitchio.ext.eventsub.EventSubWSClient
 
     """
+
     pass
 
 
@@ -1651,6 +1656,7 @@ class _SubscriptionTypes(metaclass=_SubTypesMeta):
 
 SubscriptionTypes = _SubscriptionTypes()
 
+
 class TransportType(Enum):
     webhook = "webhook"
     websocket = "websocket"

twitchio/ext/eventsub/websocket.py

@@ -18,22 +18,25 @@
     "notification": models.NotificationEvent,
     "revocation": models.RevokationEvent,
     "reconnect": models.ReconnectEvent,
-    "session_keepalive": models.KeepAliveEvent
+    "session_keepalive": models.KeepAliveEvent,
 }
 _messages = Union[models.NotificationEvent, models.RevokationEvent, models.ReconnectEvent, models.KeepAliveEvent]
 
 
 class _Subscription:
     __slots__ = "event", "condition", "token", "subscription_id", "cost"
+
     def __init__(self, event_type: Tuple[str, int, Type[models.EventData]], condition: Dict[str, str], token: str):
         self.event = event_type
         self.condition = condition
         self.token = token
         self.subscription_id: Optional[str] = None
         self.cost: Optional[int] = None
 
+
 _T = TypeVar("_T")
 
+
 class _WakeupList(list, Generic[_T]):
     def __init__(self, *args):
         super().__init__(*args)
@@ -45,15 +48,15 @@ def _wakeup_append(self, obj: _T) -> None:
             loop = asyncio.get_running_loop()
             for cb in self._append_waiters:
                 loop.create_task(cb(obj))
-        except: # don't wake the waiters if theres no loop
+        except:  # don't wake the waiters if theres no loop
             pass
 
     def _wakeup_pop(self, obj: _T) -> None:
         try:
             loop = asyncio.get_running_loop()
             for cb in self._pop_waiters:
                 loop.create_task(cb(obj))
-        except: # don't wake the waiters
+        except:  # don't wake the waiters
             pass
 
     def append(self, obj: _T) -> None:
@@ -93,7 +96,7 @@ def __init__(self, client: Client, http: http.EventSubHTTP):
         self._timeout: Optional[int] = None
         self._session_id: Optional[str] = None
 
-        self.remaining_slots: int = 100 # default to 100
+        self.remaining_slots: int = 100  # default to 100
 
     @property
     def session_id(self) -> Optional[str]:
@@ -107,7 +110,9 @@ async def _subscribe(self, obj: _Subscription) -> dict:
         resp = await self._http.create_websocket_subscription(obj.event, obj.condition, self._session_id, obj.token)
         data = resp["data"][0]
         cost = data["cost"]
-        self.remaining_slots = resp["total_cost"] - resp["max_total_cost"] # FIXME: twitch is pretty vague about these, check back on their values later
+        self.remaining_slots = (
+            resp["total_cost"] - resp["max_total_cost"]
+        )  # FIXME: twitch is pretty vague about these, check back on their values later
         obj.cost = cost
 
         return data
@@ -122,7 +127,7 @@ async def _wakeup_and_connect(self, obj: _Subscription):
 
     async def connect(self, reconnect_url: Optional[str] = None):
         if not self._subscription_pool:
-            return # TODO: should this raise?
+            return  # TODO: should this raise?
 
         async with aiohttp.ClientSession() as session:
             sock = self._sock = await session.ws_connect(reconnect_url or self.URL)
@@ -137,18 +142,20 @@ async def connect(self, reconnect_url: Optional[str] = None):
 
         self._pump_task = self.client.loop.create_task(self.pump())
 
-        if reconnect_url: # don't resubscribe to events
+        if reconnect_url:  # don't resubscribe to events
             return
 
         for sub in self._subscription_pool:
-            await self._subscribe(sub) # TODO: how do I return this to the end user (do I bother?)
+            await self._subscribe(sub)  # TODO: how do I return this to the end user (do I bother?)
 
     async def pump(self) -> None:
         while self.is_connected:
             try:
-                msg = await cast(aiohttp.ClientWebSocketResponse, self._sock).receive_str(timeout=self._timeout+1) # extra jitter on the timeout in case of network lag
+                msg = await cast(aiohttp.ClientWebSocketResponse, self._sock).receive_str(
+                    timeout=self._timeout + 1
+                )  # extra jitter on the timeout in case of network lag
                 if not msg:
-                    continue # TODO: should this raise?
+                    continue  # TODO: should this raise?
 
                 logger.debug("Received websocket payload: %s", msg)
                 frame: _messages = self.parse_frame(_loads(msg))
@@ -171,12 +178,16 @@ async def pump(self) -> None:
                     sock = self._sock
                     self._sock = None
                     await self.connect(frame.reconnect_url)
-                    await cast(aiohttp.ClientWebSocketResponse, sock).close(code=aiohttp.WSCloseCode.GOING_AWAY, message=b"reconnecting")
+                    await cast(aiohttp.ClientWebSocketResponse, sock).close(
+                        code=aiohttp.WSCloseCode.GOING_AWAY, message=b"reconnecting"
+                    )
                     return
 
             except asyncio.TimeoutError:
                 logger.warning(f"Websocket timed out (timeout: {self._timeout}), reconnecting")
-                await cast(aiohttp.ClientWebSocketResponse, self._sock).close(code=aiohttp.WSCloseCode.ABNORMAL_CLOSURE, message=b"timeout surpassed")
+                await cast(aiohttp.ClientWebSocketResponse, self._sock).close(
+                    code=aiohttp.WSCloseCode.ABNORMAL_CLOSURE, message=b"timeout surpassed"
+                )
                 await self.connect()
                 return
 
@@ -219,7 +230,10 @@ def subscribe_user_updated(self, user: Union[PartialUser, str, int], token: str)
         self._assign_subscription(sub)
 
     def subscribe_channel_raid(
-        self, token: str, from_broadcaster: Union[PartialUser, str, int] = None, to_broadcaster: Union[PartialUser, str, int] = None
+        self,
+        token: str,
+        from_broadcaster: Union[PartialUser, str, int] = None,
+        to_broadcaster: Union[PartialUser, str, int] = None,
     ):
         if (not from_broadcaster and not to_broadcaster) or (from_broadcaster and to_broadcaster):
             raise ValueError("Expected 1 of from_broadcaster or to_broadcaster")
@@ -239,7 +253,11 @@ def subscribe_channel_raid(
         self._assign_subscription(sub)
 
     def _subscribe_channel_points_reward(
-        self, event: Tuple[str, int, Type[models._DataType]], broadcaster: Union[PartialUser, str, int], token: str, reward_id: str = None
+        self,
+        event: Tuple[str, int, Type[models._DataType]],
+        broadcaster: Union[PartialUser, str, int],
+        token: str,
+        reward_id: str = None,
     ):
         if isinstance(broadcaster, PartialUser):
             broadcaster = broadcaster.id
@@ -261,13 +279,13 @@ def _subscribe_with_broadcaster(
         broadcaster = str(broadcaster)
         sub = _Subscription(event, {"broadcaster_user_id": broadcaster}, token)
         self._assign_subscription(sub)
-    
+
     def _subscribe_with_broadcaster_moderator(
         self,
         event: Tuple[str, int, Type[models._DataType]],
         broadcaster: Union[PartialUser, str, int],
         moderator: Union[PartialUser, str, int],
-        token: str
+        token: str,
     ):
         if isinstance(broadcaster, PartialUser):
             broadcaster = broadcaster.id
@@ -279,7 +297,6 @@ def _subscribe_with_broadcaster_moderator(
         sub = _Subscription(event, {"broadcaster_user_id": broadcaster, "moderator_user_id": moderator}, token)
         self._assign_subscription(sub)
 
-
     def subscribe_channel_bans(self, broadcaster: Union[PartialUser, str, int], token: str):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.ban, broadcaster, token)
 
@@ -310,7 +327,9 @@ def subscribe_channel_follows(self, broadcaster: Union[PartialUser, str, int], t
     def subscribe_channel_follows_v2(
         self, broadcaster: Union[PartialUser, str, int], moderator: Union[PartialUser, str, int], token: str
     ):
-        return self._subscribe_with_broadcaster_moderator(models.SubscriptionTypes.followV2, broadcaster, moderator, token)
+        return self._subscribe_with_broadcaster_moderator(
+            models.SubscriptionTypes.followV2, broadcaster, moderator, token
+        )
 
     def subscribe_channel_moderators_add(self, broadcaster: Union[PartialUser, str, int], token: str):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.channel_moderator_add, broadcaster, token)
@@ -342,27 +361,37 @@ def subscribe_channel_stream_start(self, broadcaster: Union[PartialUser, str, in
     def subscribe_channel_stream_end(self, broadcaster: Union[PartialUser, str, int], token: str):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.stream_end, broadcaster, token)
 
-    def subscribe_channel_points_reward_added(self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str):
+    def subscribe_channel_points_reward_added(
+        self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str
+    ):
         return self._subscribe_channel_points_reward(
             models.SubscriptionTypes.channel_reward_add, broadcaster, token, reward_id
         )
 
-    def subscribe_channel_points_reward_updated(self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str):
+    def subscribe_channel_points_reward_updated(
+        self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str
+    ):
         return self._subscribe_channel_points_reward(
             models.SubscriptionTypes.channel_reward_update, broadcaster, token, reward_id
         )
 
-    def subscribe_channel_points_reward_removed(self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str):
+    def subscribe_channel_points_reward_removed(
+        self, broadcaster: Union[PartialUser, str, int], reward_id: str, token: str
+    ):
         return self._subscribe_channel_points_reward(
             models.SubscriptionTypes.channel_reward_remove, broadcaster, token, reward_id
         )
 
-    def subscribe_channel_points_redeemed(self, broadcaster: Union[PartialUser, str, int], token: str, reward_id: str = None):
+    def subscribe_channel_points_redeemed(
+        self, broadcaster: Union[PartialUser, str, int], token: str, reward_id: str = None
+    ):
         return self._subscribe_channel_points_reward(
             models.SubscriptionTypes.channel_reward_redeem, broadcaster, token, reward_id
         )
 
-    def subscribe_channel_points_redeem_updated(self, broadcaster: Union[PartialUser, str, int], token: str, reward_id: str = None):
+    def subscribe_channel_points_redeem_updated(
+        self, broadcaster: Union[PartialUser, str, int], token: str, reward_id: str = None
+    ):
         return self._subscribe_channel_points_reward(
             models.SubscriptionTypes.channel_reward_redeem_updated, broadcaster, token, reward_id
         )
@@ -388,7 +417,6 @@ def subscribe_channel_prediction_lock(self, broadcaster: Union[PartialUser, str,
     def subscribe_channel_prediction_end(self, broadcaster: Union[PartialUser, str, int], token: str):
         return self._subscribe_with_broadcaster(models.SubscriptionTypes.prediction_end, broadcaster, token)
 
-    
     def subscribe_channel_shield_mode_begin(
         self, broadcaster: Union[PartialUser, str, int], moderator: Union[PartialUser, str, int], token: str
     ):