Appease Ruff TC006

Author: chillymosh

twitchio/eventsub/websockets.py

@@ -34,16 +34,6 @@
 from ..backoff import Backoff
 from ..exceptions import HTTPException, WebsocketConnectionException
 from ..models.eventsub_ import SubscriptionRevoked, create_event_instance
-from ..types_.conduits import (
-    MessageTypes,
-    MetaData,
-    NotificationMessage,
-    ReconnectMessage,
-    RevocationMessage,
-    WebsocketMessages,
-    WelcomeMessage,
-    WelcomePayload,
-)
 from ..utils import (
     MISSING,
     _from_json,  # type: ignore
@@ -54,7 +44,17 @@
 if TYPE_CHECKING:
     from ..authentication.tokens import ManagedHTTPClient
     from ..client import Client
-    from ..types_.conduits import Condition
+    from ..types_.conduits import (
+        Condition,
+        MessageTypes,
+        MetaData,
+        NotificationMessage,
+        ReconnectMessage,
+        RevocationMessage,
+        WebsocketMessages,
+        WelcomeMessage,
+        WelcomePayload,
+    )
     from ..types_.eventsub import SubscriptionResponse, _SubscriptionData
 
 
@@ -327,7 +327,7 @@ async def _listen(self) -> None:
             self._last_keepalive = datetime.datetime.now()
 
             try:
-                data: WebsocketMessages = cast(WebsocketMessages, _from_json(message.data))
+                data: WebsocketMessages = cast("WebsocketMessages", _from_json(message.data))
             except Exception:
                 logger.warning("Unable to parse JSON in eventsub websocket: <%s>", self)
                 continue
@@ -336,13 +336,13 @@ async def _listen(self) -> None:
             message_type: MessageTypes = metadata["message_type"]
 
             if message_type == "session_welcome":
-                welcome_data: WelcomeMessage = cast(WelcomeMessage, data)
+                welcome_data: WelcomeMessage = cast("WelcomeMessage", data)
 
                 await self._process_welcome(welcome_data)
 
             elif message_type == "session_reconnect":
                 logger.debug('Received "session_reconnect" message from eventsub websocket: <%s>', self)
-                reconnect_data: ReconnectMessage = cast(ReconnectMessage, data)
+                reconnect_data: ReconnectMessage = cast("ReconnectMessage", data)
 
                 await self._process_reconnect(reconnect_data)
 
@@ -352,12 +352,12 @@ async def _listen(self) -> None:
             elif message_type == "revocation":
                 logger.debug('Received "revocation" message from eventsub websocket: <%s>', self)
 
-                revocation_data: RevocationMessage = cast(RevocationMessage, data)
+                revocation_data: RevocationMessage = cast("RevocationMessage", data)
                 await self._process_revocation(revocation_data)
 
             elif message_type == "notification":
                 logger.debug('Received "notification" message from eventsub websocket: <%s>. %s', self, data)
-                notification_data: NotificationMessage = cast(NotificationMessage, data)
+                notification_data: NotificationMessage = cast("NotificationMessage", data)
 
                 try:
                     await self._process_notification(notification_data)

twitchio/utils.py

@@ -929,7 +929,7 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
 
             return func(*new_args, **new_kwargs)
 
-        return cast(F, wrapper)
+        return cast("F", wrapper)
 
     return decorator
 

twitchio/web/aio_adapter.py

@@ -37,7 +37,6 @@
 from ..eventsub.subscriptions import _SUB_MAPPING
 from ..exceptions import HTTPException
 from ..models.eventsub_ import SubscriptionRevoked, create_event_instance
-from ..types_.eventsub import EventSubHeaders
 from ..utils import _from_json, parse_timestamp  # type: ignore
 from .utils import MESSAGE_TYPES, BaseAdapter, FetchTokenPayload, verify_message
 
@@ -47,6 +46,7 @@
 
     from ..authentication import AuthorizationURLPayload, UserTokenPayload
     from ..client import Client
+    from ..types_.eventsub import EventSubHeaders
 
 
 __all__ = ("AiohttpAdapter",)
@@ -221,7 +221,7 @@ async def run(self, host: str | None = None, port: int | None = None) -> None:
         self._runner_task = asyncio.create_task(site.start(), name=f"twitchio-web-adapter:{self.__class__.__qualname__}")
 
     async def eventsub_callback(self, request: web.Request) -> web.Response:
-        headers: EventSubHeaders = cast(EventSubHeaders, request.headers)
+        headers: EventSubHeaders = cast("EventSubHeaders", request.headers)
         msg_type: str | None = headers.get("Twitch-Eventsub-Message-Type")
 
         if not msg_type or msg_type not in MESSAGE_TYPES:

twitchio/web/starlette_adapter.py

@@ -40,7 +40,6 @@
 from ..eventsub.subscriptions import _SUB_MAPPING
 from ..exceptions import HTTPException
 from ..models.eventsub_ import SubscriptionRevoked, create_event_instance
-from ..types_.eventsub import EventSubHeaders
 from ..utils import _from_json, parse_timestamp  # type: ignore
 from .utils import MESSAGE_TYPES, BaseAdapter, FetchTokenPayload, verify_message
 
@@ -50,6 +49,7 @@
 
     from ..authentication import AuthorizationURLPayload, UserTokenPayload
     from ..client import Client
+    from ..types_.eventsub import EventSubHeaders
 
 
 __all__ = ("StarletteAdapter",)
@@ -226,7 +226,7 @@ async def run(self, host: str | None = None, port: int | None = None) -> None:
         self._runner_task = asyncio.create_task(server.serve(), name=f"twitchio-web-adapter:{self.__class__.__qualname__}")
 
     async def eventsub_callback(self, request: Request) -> Response:
-        headers: EventSubHeaders = cast(EventSubHeaders, request.headers)
+        headers: EventSubHeaders = cast("EventSubHeaders", request.headers)
         msg_type: str | None = headers.get("Twitch-Eventsub-Message-Type")
 
         if not msg_type or msg_type not in MESSAGE_TYPES: