Add authorisation_by_user endpoint support

Author: chillymosh

docs/getting-started/changelog.rst

@@ -10,8 +10,12 @@ Changelog
 ======
 - twitchio
     - Additions
-        - Added - :func:`~twitchio.PartialUser.fetch_hype_train_status`
-        - Added - :func:`~twitchio.PartialUser.fetch_stream`
+        - Added - :class:`~twitchio.UserAuthorisation` model
+        - Added - :func:`~twitchio.Client.fetch_auth_by_users` to :class:`~twitchio.Client`
+        - Added - :func:`~twitchio.PartialUser.fetch_auth` to :class:`~twitchio.PartialUser`
+        - Added - :func:`~twitchio.PartialUser.fetch_stream` to :class:`~twitchio.PartialUser` as a helper method.
+        - Added - :func:`~twitchio.PartialUser.fetch_hype_train_status` to :class:`~twitchio.PartialUser`. 
+            This replaces :func:`~twitchio.PartialUser.fetch_hype_train_events` which has been deprecated
 
 
 3.1.0

docs/references/helix/helix_models.rst

@@ -367,6 +367,11 @@
 .. autoclass:: twitchio.VideoMarkers
     :members:
 
+.. attributetable:: twitchio.UserAuthorisation
+
+.. autoclass:: twitchio.UserAuthorisation
+    :members:
+
 .. attributetable:: twitchio.UserSubscription
 
 .. autoclass:: twitchio.UserSubscription

twitchio/client.py

@@ -44,7 +44,7 @@
 from .models.games import Game
 from .models.teams import Team
 from .payloads import EventErrorPayload, WebsocketSubscriptionData
-from .user import ActiveExtensions, Extension, PartialUser, User
+from .user import ActiveExtensions, Extension, PartialUser, User, UserAuthorisation
 from .utils import MISSING, EventWaiter, clamp, unwrap_function
 from .web import AiohttpAdapter, has_starlette
 from .web.utils import BaseAdapter
@@ -2205,6 +2205,32 @@ async def update_entitlements(
         data = await self._http.patch_drop_entitlements(ids=ids, fulfillment_status=fulfillment_status, token_for=token_for)
         return [EntitlementStatus(d) for d in data["data"]]
 
+    async def fetch_auth_by_users(self, user_ids: list[str]) -> list[UserAuthorisation]:
+        """|coro|
+        Fetches authentication information for one or more users, up to 10.
+        This uses the app token associated with the Client ID and Secret provided.
+
+        Parameters
+        ----------
+        user_ids: list[str]
+            A list of user IDs to fetch authentication information for.
+            You may only fetch a maximum of `10` user IDs at a time.
+
+        Returns
+        -------
+        list[UserAuthorisation]
+
+        Raises
+        ------
+        ValueError
+            You may only fetch a maximum of ``10`` user IDs at a time.
+        """
+        if len(user_ids) > 10:
+            raise ValueError("You may only fetch a maximum of 10 user IDs at a time.")
+
+        data = await self._http.get_auth_by_user(user_ids=user_ids)
+        return [UserAuthorisation(u, http=self._http) for u in data["data"]]
+
     async def subscribe_websocket(
         self,
         payload: SubscriptionPayload,

twitchio/http.py

@@ -81,6 +81,7 @@
     from .types_.responses import (
         AddBlockedTermResponse,
         AdScheduleResponse,
+        AuthorizationByUserResponse,
         AutomodSettingsResponse,
         BannedUsersResponseData,
         BanUserResponse,
@@ -2510,6 +2511,11 @@ async def put_user(self, token_for: str, description: str | None) -> UpdateUserR
         route: Route = Route("PUT", "users", params=params, token_for=token_for)
         return await self.request_json(route)
 
+    async def get_auth_by_user(self, user_ids: list[str]) -> AuthorizationByUserResponse:
+        params = {"user_id": user_ids}
+        route: Route = Route("GET", "authorization/users", params=params)
+        return await self.request_json(route)
+
     def get_user_block_list(
         self, broadcaster_id: str | int, token_for: str, first: int = 20, max_results: int | None = None
     ) -> HTTPAsyncIterator[PartialUser]:

twitchio/types_/responses.py

@@ -30,6 +30,8 @@
 __all__ = (
     "AdScheduleResponse",
     "AdScheduleResponseData",
+    "AuthorizationByUserResponse",
+    "AuthorizationByUserResponseData",
     "AuthorizationURLResponse",
     "BitsLeaderboardResponse",
     "ChannelChatBadgesResponseData",
@@ -1663,6 +1665,18 @@ class UpdateUserResponse(TypedDict):
     data: list[UsersResponseData]
 
 
+class AuthorizationByUserResponseData(TypedDict):
+    user_id: str
+    user_name: str
+    user_login: str
+    scopes: list[str]
+    has_authorized: bool
+
+
+class AuthorizationByUserResponse(TypedDict):
+    data: list[AuthorizationByUserResponseData]
+
+
 class UserBlockListResponseData(TypedDict):
     user_id: str
     user_login: str

twitchio/user.py

@@ -37,6 +37,7 @@
     import datetime
 
     from twitchio.types_.responses import (
+        AuthorizationByUserResponseData,
         UserActiveExtensionsResponseData,
         UserExtensionsResponseData,
         UserPanelComponentItem,
@@ -75,7 +76,7 @@
     from .models.subscriptions import BroadcasterSubscriptions, UserSubscription
     from .models.teams import ChannelTeam
 
-__all__ = ("ActiveExtensions", "Extension", "PartialUser", "User")
+__all__ = ("ActiveExtensions", "Extension", "PartialUser", "User", "UserAuthorisation")
 
 
 class PartialUser:
@@ -3211,6 +3212,20 @@ async def update(self, description: str | None = None) -> User:
         data = await self._http.put_user(token_for=self.id, description=description)
         return User(data["data"][0], http=self._http)
 
+    async def fetch_auth(self) -> UserAuthorisation:
+        """|coro|
+
+        Fetches the user's authentication information.
+
+        Returns
+        -------
+        UserAuthorisation
+            UserAuthorisation object.
+        """
+
+        data = await self._http.get_auth_by_user(user_ids=[self.id])
+        return UserAuthorisation(data["data"][0], http=self._http)
+
     def fetch_block_list(self, *, first: int = 20, max_results: int | None = None) -> HTTPAsyncIterator[PartialUser]:
         """|aiter|
 
@@ -4061,3 +4076,29 @@ async def follow_info(self) -> ChannelFollowerEvent | None:
             user_id=self, broadcaster_id=self.channel, token_for=self.channel, max_results=1
         )
         return await anext(data.followers, None)
+
+
+class UserAuthorisation:
+    """A class that contains authorisation information for a user against a Client ID / Twitch application.
+
+    Attributes
+    -----------
+    user: PartialUser
+        The user having authorised checked.
+    scopes: Scopes
+        The scopes the user has granted to the Client ID / application.
+    authorised: bool
+        Whether the user has authorised the Client ID / application.
+    """
+
+    __slots__ = ("authorised", "scopes", "user")
+
+    def __init__(self, data: AuthorizationByUserResponseData, *, http: HTTPClient) -> None:
+        from .authentication import Scopes
+
+        self.user: PartialUser = PartialUser(data["user_id"], data["user_login"], data["user_name"], http=http)
+        self.scopes: Scopes = Scopes(data["scopes"])
+        self.authorised: bool = bool(data["has_authorized"])
+
+    def __repr__(self) -> str:
+        return f"<UserAuthorisation user={self.user} authorised={self.authorised}>"