Make Web Adapters Generic and accept a Client or any derivatives.

Author: EvieePy

twitchio/client.py

@@ -29,7 +29,7 @@
 import math
 from collections import defaultdict
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Self, Unpack, overload
+from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Self, Unpack, cast, overload
 
 from .authentication import ManagedHTTPClient, Scopes, UserTokenPayload
 from .eventsub.enums import SubscriptionType
@@ -149,12 +149,14 @@ def __init__(
             msg = "If you require the StarletteAdapter please install the required packages: 'pip install twitchio[starlette]'."
             logger.warning(msg)
 
-        adapter: BaseAdapter | type[BaseAdapter] = options.get("adapter", AiohttpAdapter)
+        adapter: BaseAdapter[Any] | type[BaseAdapter[Any]] | type[AiohttpAdapter[Self]] = options.get(  # type: ignore
+            "adapter", AiohttpAdapter
+        )
         if isinstance(adapter, BaseAdapter):
             adapter.client = self
             self._adapter = adapter
         else:
-            self._adapter = adapter()
+            self._adapter = cast("AiohttpAdapter[Self]", adapter())
             self._adapter.client = self
 
         # Own Client User. Set in login...
@@ -178,12 +180,12 @@ def __init__(
         self._setup_called = False
 
     @property
-    def adapter(self) -> BaseAdapter:
+    def adapter(self) -> BaseAdapter[Any]:
         """Property returning the :class:`~twitchio.AiohttpAdapter` or :class:`~twitchio.StarlettepAdapter` the bot is
         currently running."""
         return self._adapter
 
-    async def set_adapter(self, adapter: BaseAdapter) -> None:
+    async def set_adapter(self, adapter: BaseAdapter[Any]) -> None:
         """|coro|
 
         Method which sets and starts a new web adapter which inherits from either :class:`~twitchio.AiohttpAdapter` or

twitchio/types_/options.py

@@ -43,7 +43,7 @@ class ClientOptions(TypedDict, total=False):
     redirect_uri: str | None
     scopes: Scopes | None
     session: aiohttp.ClientSession | None
-    adapter: NotRequired[BaseAdapter]
+    adapter: NotRequired[BaseAdapter[Any]]
     fetch_client_user: NotRequired[bool]
 
 

twitchio/web/aio_adapter.py

@@ -28,7 +28,7 @@
 import datetime
 import logging
 from collections import deque
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 from urllib.parse import unquote_plus
 
 from aiohttp import web
@@ -54,10 +54,11 @@
 __all__ = ("AiohttpAdapter",)
 
 
+BT = TypeVar("BT", bound="Client")
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-class AiohttpAdapter(BaseAdapter, web.Application):
+class AiohttpAdapter(BaseAdapter[BT], web.Application):
     """The AiohttpAdapter for OAuth and Webhook based EventSub.
 
     This adapter uses ``aiohttp`` which is a base dependency and should be installed and available with Twitchio.
@@ -141,7 +142,7 @@ def __init__(self) -> None:
                 super().__init__(adapter=adapter)
     """
 
-    client: Client
+    client: BT
 
     def __init__(
         self,
@@ -194,7 +195,7 @@ def __init__(
         self._responded: deque[str] = deque(maxlen=5000)
         self._running: bool = False
 
-    def __init_subclass__(cls: type[AiohttpAdapter]) -> None:
+    def __init_subclass__(cls: type[AiohttpAdapter[BT]]) -> None:
         return
 
     def __repr__(self) -> str:

twitchio/web/starlette_adapter.py

@@ -28,7 +28,7 @@
 import datetime
 import logging
 from collections import deque
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 from urllib.parse import unquote_plus
 
 import uvicorn
@@ -57,10 +57,11 @@
 __all__ = ("StarletteAdapter",)
 
 
+BT = TypeVar("BT", bound="Client")
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-class StarletteAdapter(BaseAdapter, Starlette):
+class StarletteAdapter(BaseAdapter[BT], Starlette):
     """The StarletteAdapter for OAuth and Webhook based EventSub.
 
     This adapter uses ``starlette`` which is an optional dependency and needs to be installed.
@@ -153,7 +154,7 @@ def __init__(self) -> None:
                 super().__init__(adapter=adapter)
     """
 
-    client: Client
+    client: BT
 
     def __init__(
         self,

twitchio/web/utils.py

@@ -28,7 +28,7 @@
 import hashlib
 import hmac
 import logging
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 from aiohttp import web
 
@@ -45,6 +45,7 @@
     from ..types_.eventsub import EventSubHeaders
 
 
+BT = TypeVar("BT", bound="Client")
 logger: logging.Logger = logging.getLogger(__name__)
 
 
@@ -82,8 +83,8 @@ def __init__(
         self.exception = exception
 
 
-class BaseAdapter(abc.ABC):
-    client: Client
+class BaseAdapter(abc.ABC, Generic[BT]):
+    client: BT
     _runner_task: asyncio.Task[None] | None
     _eventsub_secret: str | None
     _running: bool