🚧 OMG THIS WORKS

Author: Paillat-dev

discord/asset.py

@@ -30,6 +30,7 @@
 from typing import TYPE_CHECKING, Any, Literal
 
 import yarl
+from typing_extensions import Final, override
 
 from . import utils
 from .errors import DiscordException, InvalidArgument
@@ -39,6 +40,7 @@
 if TYPE_CHECKING:
     ValidStaticFormatTypes = Literal["webp", "jpeg", "jpg", "png"]
     ValidAssetFormatTypes = Literal["webp", "jpeg", "jpg", "png", "gif"]
+    from .state import ConnectionState
 
 VALID_STATIC_FORMATS = frozenset({"jpeg", "jpg", "webp", "png"})
 VALID_ASSET_FORMATS = VALID_STATIC_FORMATS | {"gif"}
@@ -49,7 +51,7 @@
 
 class AssetMixin:
     url: str
-    _state: Any | None
+    _state: ConnectionState | None
 
     async def read(self) -> bytes:
         """|coro|
@@ -77,7 +79,9 @@ async def read(self) -> bytes:
 
     async def save(
         self,
-        fp: str | bytes | os.PathLike | io.BufferedIOBase,
+        fp: (
+            str | bytes | os.PathLike | io.BufferedIOBase
+        ),  # pyright: ignore [reportMissingTypeArgument]
         *,
         seek_begin: bool = True,
     ) -> int:
@@ -117,7 +121,7 @@ async def save(
                 fp.seek(0)
             return written
         else:
-            with open(fp, "wb") as f:
+            with open(fp, "wb") as f:  # pyright: ignore [reportUnknownArgumentType]
                 return f.write(data)
 
 
@@ -154,16 +158,23 @@ class Asset(AssetMixin):
         "_key",
     )
 
-    BASE = "https://cdn.discordapp.com"
+    BASE: Final = "https://cdn.discordapp.com"
 
-    def __init__(self, state, *, url: str, key: str, animated: bool = False):
-        self._state = state
-        self._url = url
-        self._animated = animated
-        self._key = key
+    def __init__(
+        self,
+        state: ConnectionState | None,
+        *,
+        url: str,
+        key: str,
+        animated: bool = False,
+    ):
+        self._state: ConnectionState | None = state
+        self._url: str = url
+        self._animated: bool = animated
+        self._key: str = key
 
     @classmethod
-    def _from_default_avatar(cls, state, index: int) -> Asset:
+    def _from_default_avatar(cls, state: ConnectionState, index: int) -> Asset:
         return cls(
             state,
             url=f"{cls.BASE}/embed/avatars/{index}.png",
@@ -172,7 +183,7 @@ def _from_default_avatar(cls, state, index: int) -> Asset:
         )
 
     @classmethod
-    def _from_avatar(cls, state, user_id: int, avatar: str) -> Asset:
+    def _from_avatar(cls, state: ConnectionState, user_id: int, avatar: str) -> Asset:
         animated = avatar.startswith("a_")
         format = "gif" if animated else "png"
         return cls(
@@ -184,7 +195,10 @@ def _from_avatar(cls, state, user_id: int, avatar: str) -> Asset:
 
     @classmethod
     def _from_avatar_decoration(
-        cls, state, user_id: int, avatar_decoration: str
+        cls,
+        state: ConnectionState,
+        user_id: int,
+        avatar_decoration: str,  # pyright: ignore [reportUnusedParameter]
     ) -> Asset:
         animated = avatar_decoration.startswith("a_")
         endpoint = (
@@ -201,7 +215,7 @@ def _from_avatar_decoration(
 
     @classmethod
     def _from_guild_avatar(
-        cls, state, guild_id: int, member_id: int, avatar: str
+        cls, state: ConnectionState, guild_id: int, member_id: int, avatar: str
     ) -> Asset:
         animated = avatar.startswith("a_")
         format = "gif" if animated else "png"
@@ -214,7 +228,7 @@ def _from_guild_avatar(
 
     @classmethod
     def _from_guild_banner(
-        cls, state, guild_id: int, member_id: int, banner: str
+        cls, state: ConnectionState, guild_id: int, member_id: int, banner: str
     ) -> Asset:
         animated = banner.startswith("a_")
         format = "gif" if animated else "png"
@@ -226,7 +240,9 @@ def _from_guild_banner(
         )
 
     @classmethod
-    def _from_icon(cls, state, object_id: int, icon_hash: str, path: str) -> Asset:
+    def _from_icon(
+        cls, state: ConnectionState, object_id: int, icon_hash: str, path: str
+    ) -> Asset:
         return cls(
             state,
             url=f"{cls.BASE}/{path}-icons/{object_id}/{icon_hash}.png?size=1024",
@@ -235,7 +251,9 @@ def _from_icon(cls, state, object_id: int, icon_hash: str, path: str) -> Asset:
         )
 
     @classmethod
-    def _from_cover_image(cls, state, object_id: int, cover_image_hash: str) -> Asset:
+    def _from_cover_image(
+        cls, state: ConnectionState, object_id: int, cover_image_hash: str
+    ) -> Asset:
         return cls(
             state,
             url=f"{cls.BASE}/app-assets/{object_id}/store/{cover_image_hash}.png?size=1024",
@@ -244,7 +262,9 @@ def _from_cover_image(cls, state, object_id: int, cover_image_hash: str) -> Asse
         )
 
     @classmethod
-    def _from_guild_image(cls, state, guild_id: int, image: str, path: str) -> Asset:
+    def _from_guild_image(
+        cls, state: ConnectionState, guild_id: int, image: str, path: str
+    ) -> Asset:
         animated = False
         format = "png"
         if path == "banners":
@@ -259,7 +279,9 @@ def _from_guild_image(cls, state, guild_id: int, image: str, path: str) -> Asset
         )
 
     @classmethod
-    def _from_guild_icon(cls, state, guild_id: int, icon_hash: str) -> Asset:
+    def _from_guild_icon(
+        cls, state: ConnectionState, guild_id: int, icon_hash: str
+    ) -> Asset:
         animated = icon_hash.startswith("a_")
         format = "gif" if animated else "png"
         return cls(
@@ -270,7 +292,7 @@ def _from_guild_icon(cls, state, guild_id: int, icon_hash: str) -> Asset:
         )
 
     @classmethod
-    def _from_sticker_banner(cls, state, banner: int) -> Asset:
+    def _from_sticker_banner(cls, state: ConnectionState, banner: int) -> Asset:
         return cls(
             state,
             url=f"{cls.BASE}/app-assets/710982414301790216/store/{banner}.png",
@@ -279,7 +301,9 @@ def _from_sticker_banner(cls, state, banner: int) -> Asset:
         )
 
     @classmethod
-    def _from_user_banner(cls, state, user_id: int, banner_hash: str) -> Asset:
+    def _from_user_banner(
+        cls, state: ConnectionState, user_id: int, banner_hash: str
+    ) -> Asset:
         animated = banner_hash.startswith("a_")
         format = "gif" if animated else "png"
         return cls(
@@ -291,7 +315,7 @@ def _from_user_banner(cls, state, user_id: int, banner_hash: str) -> Asset:
 
     @classmethod
     def _from_scheduled_event_image(
-        cls, state, event_id: int, cover_hash: str
+        cls, state: ConnectionState, event_id: int, cover_hash: str
     ) -> Asset:
         return cls(
             state,
@@ -300,24 +324,29 @@ def _from_scheduled_event_image(
             animated=False,
         )
 
+    @override
     def __str__(self) -> str:
         return self._url
 
     def __len__(self) -> int:
         return len(self._url)
 
+    @override
     def __repr__(self):
         shorten = self._url.replace(self.BASE, "")
         return f"<Asset url={shorten!r}>"
 
-    def __eq__(self, other):
+    @override
+    def __eq__(self, other: Any):  # pyright: ignore [reportExplicitAny]
         return isinstance(other, Asset) and self._url == other._url
 
+    @override
     def __hash__(self):
         return hash(self._url)
 
     @property
-    def url(self) -> str:
+    @override
+    def url(self) -> str:  # pyright: ignore [reportIncompatibleVariableOverride]
         """Returns the underlying URL of the asset."""
         return self._url
 

discord/client.py

@@ -35,7 +35,7 @@
 
 import aiohttp
 
-from . import utils
+from . import models, utils
 from .activity import ActivityTypes, BaseActivity, create_activity
 from .appinfo import AppInfo, PartialAppInfo
 from .application_role_connection import ApplicationRoleConnectionMetadata
@@ -1840,7 +1840,7 @@ async def fetch_user(self, user_id: int, /) -> User:
         :exc:`HTTPException`
             Fetching the user failed.
         """
-        data = await self.http.get_user(user_id)
+        data: models.User = await self.http.get_user(user_id)
         return User(state=self._connection, data=data)
 
     async def fetch_channel(

discord/gateway.py

@@ -35,8 +35,12 @@
 import traceback
 import zlib
 from collections import deque, namedtuple
+from typing import Any
 
 import aiohttp
+from pydantic import BaseModel
+
+from discord import models
 
 from . import utils
 from .activity import BaseActivity
@@ -548,11 +552,20 @@ async def received_message(self, msg, /):
             )
 
         try:
-            func = self._discord_parsers[event]
+            func: Any = self._discord_parsers[event]
         except KeyError:
             _log.debug("Unknown event %s.", event)
         else:
-            func(data)
+            if hasattr(func, "_supports_model") and issubclass(
+                func._supports_model, models.gateway.GatewayEvent
+            ):
+                func(
+                    func._supports_model(
+                        **msg
+                    ).d  # pyright: ignore [reportUnknownMemberType, reportAttributeAccessIssue]
+                )
+            else:
+                func(data)
 
         # remove the dispatched listeners
         removed = []

discord/http.py

@@ -33,8 +33,9 @@
 from urllib.parse import quote as _uriquote
 
 import aiohttp
+from typing_extensions import overload
 
-from . import __version__, utils
+from . import __version__, models, utils
 from .errors import (
     DiscordServerError,
     Forbidden,
@@ -52,6 +53,8 @@
 if TYPE_CHECKING:
     from types import TracebackType
 
+    from pydantic import BaseModel
+
     from .enums import AuditLogAction, InteractionResponseType
     from .file import File
     from .types import (
@@ -87,9 +90,11 @@
     T = TypeVar("T")
     BE = TypeVar("BE", bound=BaseException)
     MU = TypeVar("MU", bound="MaybeUnlock")
-    Response = Coroutine[Any, Any, T]
+
+    Response = Coroutine[Any, Any, T]  # pyright: ignore [reportExplicitAny]
 
 API_VERSION: int = 10
+BM = TypeVar("BM", bound=type["BaseModel"])
 
 
 async def json_or_text(response: aiohttp.ClientResponse) -> dict[str, Any] | str:
@@ -157,7 +162,7 @@ def __exit__(
 
 # For some reason, the Discord voice websocket expects this header to be
 # completely lowercase while aiohttp respects spec and does it as case-insensitive
-aiohttp.hdrs.WEBSOCKET = "websocket"  # type: ignore
+aiohttp.hdrs.WEBSOCKET = "websocket"  # type: ignore # pyright: ignore [reportAttributeAccessIssue]
 
 
 class HTTPClient:
@@ -215,14 +220,36 @@ async def ws_connect(self, url: str, *, compress: int = 0) -> Any:
 
         return await self.__session.ws_connect(url, **kwargs)
 
+    @overload
     async def request(
         self,
         route: Route,
         *,
         files: Sequence[File] | None = None,
         form: Iterable[dict[str, Any]] | None = None,
+        model: None = None,
         **kwargs: Any,
-    ) -> Any:
+    ) -> Any: ...
+
+    @overload
+    async def request(
+        self,
+        route: Route,
+        *,
+        files: None = ...,
+        form: None = ...,
+        model: BM,
+        **kwargs: Any,
+    ) -> BM: ...
+    async def request(
+        self,
+        route: Route,
+        *,
+        files: Sequence[File] | None = None,
+        form: Iterable[dict[str, Any]] | None = None,
+        model: BM | None = None,
+        **kwargs: Any,
+    ) -> Any | BM:
         bucket = route.bucket
         method = route.method
         url = route.url
@@ -318,6 +345,10 @@ async def request(
                         # the request was successful so just return the text/json
                         if 300 > response.status >= 200:
                             _log.debug("%s %s has received %s", method, url, data)
+                            if model:
+                                return model(
+                                    **data
+                                )  # pyright: ignore [reportCallIssue]
                             return data
 
                         # we are being rate limited
@@ -409,7 +440,7 @@ async def close(self) -> None:
 
     # login management
 
-    async def static_login(self, token: str) -> user.User:
+    async def static_login(self, token: str) -> models.User:
         # Necessary to get aiohttp to stop complaining about session creation
         self.__session = aiohttp.ClientSession(
             connector=self.connector, ws_response_class=DiscordClientWebSocketResponse
@@ -418,7 +449,7 @@ async def static_login(self, token: str) -> user.User:
         self.token = token
 
         try:
-            data = await self.request(Route("GET", "/users/@me"))
+            data = await self.request(Route("GET", "/users/@me"), model=models.User)
         except HTTPException as exc:
             self.token = old_token
             if exc.status == 401:
@@ -3173,5 +3204,7 @@ async def get_bot_gateway(
             value = "{0}?encoding={1}&v={2}"
         return data["shards"], value.format(data["url"], encoding, API_VERSION)
 
-    def get_user(self, user_id: Snowflake) -> Response[user.User]:
-        return self.request(Route("GET", "/users/{user_id}", user_id=user_id))
+    def get_user(self, user_id: Snowflake) -> Response[models.User]:
+        return self.request(
+            Route("GET", "/users/{user_id}", user_id=user_id), model=models.User
+        )