New PartialUser following methods

Add deprecation warnings for follower endpoints

Add new methods and models

Author: chillymosh

docs/changelog.rst

@@ -3,6 +3,17 @@
 Master
 ======
 - TwitchIO
+    - Additions
+        - Added the new follower / followed endpoints for :class:`~twitchio.PartialUser`:
+            - :func:`~twitchio.PartialUser.fetch_channel_followers`
+            - :func:`~twitchio.PartialUser.fetch_channel_following`
+            - :func:`~twitchio.PartialUser.fetch_channel_follower_count`
+            - :func:`~twitchio.PartialUser.fetch_channel_following_count`
+        - The deprecated methods have had warnings added in the docs.
+        - New models for the new methods have been added:
+            - :class:`~twitchio.ChannelFollowerEvent`
+            - :class:`~twitchio.ChannelFollowingEvent`
+
     - Bug fixes
         - Fix IndexError when getting prefix when empty message is sent in a reply.
 

docs/reference.rst

@@ -74,6 +74,22 @@ ChannelEmote
     :members:
     :inherited-members:
 
+ChannelFollowerEvent
+---------------------
+.. attributetable:: ChannelFollowerEvent
+
+.. autoclass:: ChannelFollowerEvent
+    :members:
+    :inherited-members:
+
+ChannelFollowingEvent
+---------------------
+.. attributetable:: ChannelFollowingEvent
+
+.. autoclass:: ChannelFollowingEvent
+    :members:
+    :inherited-members:
+
 ChannelInfo
 ------------
 .. attributetable:: ChannelInfo

twitchio/http.py

@@ -1178,3 +1178,67 @@ async def get_channel_charity_campaigns(self, broadcaster_id: str, token: str):
         return await self.request(
             Route("GET", "charity/campaigns", query=[("broadcaster_id", broadcaster_id)], token=token)
         )
+
+    async def get_channel_followers(
+        self,
+        token: str,
+        broadcaster_id: str,
+        user_id: Optional[int] = None,
+    ):
+        query = [("broadcaster_id", broadcaster_id)]
+
+        if user_id is not None:
+            query.append(("user_id", str(user_id)))
+
+        return await self.request(
+            Route(
+                "GET",
+                "channels/followers",
+                query=query,
+                token=token,
+            )
+        )
+
+    async def get_channel_followed(
+        self,
+        token: str,
+        user_id: str,
+        broadcaster_id: Optional[int] = None,
+    ):
+        query = [("user_id", user_id)]
+
+        if broadcaster_id is not None:
+            query.append(("broadcaster_id", str(broadcaster_id)))
+
+        return await self.request(
+            Route(
+                "GET",
+                "channels/followed",
+                query=query,
+                token=token,
+            )
+        )
+
+    async def get_channel_follower_count(self, broadcaster_id: str, token: Optional[str] = None):
+        return await self.request(
+            Route(
+                "GET",
+                "channels/followers",
+                query=[("broadcaster_id", broadcaster_id)],
+                token=token,
+            ),
+            full_body=True,
+            paginate=False,
+        )
+
+    async def get_channel_followed_count(self, token: str, user_id: str):
+        return await self.request(
+            Route(
+                "GET",
+                "channels/followed",
+                query=[("user_id", user_id)],
+                token=token,
+            ),
+            full_body=True,
+            paginate=False,
+        )

twitchio/models.py

@@ -82,6 +82,8 @@
     "ContentClassificationLabel",
     "CharityValues",
     "CharityCampaign",
+    "ChannelFollowerEvent",
+    "ChannelFollowingEvent",
 )
 
 
@@ -510,6 +512,60 @@ def __repr__(self):
         return f"<FollowEvent from_user={self.from_user} to_user={self.to_user} followed_at={self.followed_at}>"
 
 
+class ChannelFollowerEvent:
+    """
+    Represents a ChannelFollowEvent Event.
+
+    Attributes
+    -----------
+    user: Union[:class:`~twitchio.User`, :class:`~twitchio.PartialUser`]
+        The user that followed another user.
+    followed_at: :class:`datetime.datetime`
+        When the follow happened.
+    """
+
+    __slots__ = "user", "followed_at"
+
+    def __init__(
+        self,
+        http: "TwitchHTTP",
+        data: dict,
+    ):
+        self.user: Union[User, PartialUser] = PartialUser(http, data["user_id"], data["user_login"])
+        self.followed_at = parse_timestamp(data["followed_at"])
+
+    def __repr__(self):
+        return f"<ChannelFollowerEvent user={self.user} followed_at={self.followed_at}>"
+
+
+class ChannelFollowingEvent:
+    """
+    Represents a ChannelFollowEvent Event.
+
+    Attributes
+    -----------
+    broadcaster: Union[:class:`~twitchio.User`, :class:`~twitchio.PartialUser`]
+        The user that is following another user.
+    followed_at: :class:`datetime.datetime`
+        When the follow happened.
+    """
+
+    __slots__ = "broadcaster", "followed_at"
+
+    def __init__(
+        self,
+        http: "TwitchHTTP",
+        data: dict,
+    ):
+        self.broadcaster: Union[User, PartialUser] = PartialUser(
+            http, data["broadcaster_id"], data["broadcaster_login"]
+        )
+        self.followed_at = parse_timestamp(data["followed_at"])
+
+    def __repr__(self):
+        return f"<ChannelFollowerEvent user={self.broadcaster} followed_at={self.followed_at}>"
+
+
 class SubscriptionEvent:
     """
     Represents a Subscription Event

twitchio/user.py

@@ -28,15 +28,25 @@
 from typing import TYPE_CHECKING, List, Optional, Union, Tuple, Dict
 
 from .enums import BroadcasterTypeEnum, UserTypeEnum
-from .errors import HTTPException, Unauthorized
+from .errors import HTTPException
 from .rewards import CustomReward
 from .utils import parse_timestamp
 
 
 if TYPE_CHECKING:
     from .http import TwitchHTTP
     from .channel import Channel
-    from .models import BitsLeaderboard, Clip, ExtensionBuilder, Tag, FollowEvent, Prediction, CharityCampaign
+    from .models import (
+        BitsLeaderboard,
+        Clip,
+        ExtensionBuilder,
+        Tag,
+        FollowEvent,
+        Prediction,
+        CharityCampaign,
+        ChannelFollowerEvent,
+        ChannelFollowingEvent,
+    )
 __all__ = (
     "PartialUser",
     "BitLeaderboardUser",
@@ -491,6 +501,10 @@ async def fetch_following(self, token: Optional[str] = None) -> List["FollowEven
 
         Fetches a list of users that this user is following.
 
+        .. warning::
+
+            The endpoint this method uses has been deprecated by Twitch and is no longer available.
+
         Parameters
         -----------
         token: Optional[:class:`str`]
@@ -510,6 +524,10 @@ async def fetch_followers(self, token: Optional[str] = None):
 
         Fetches a list of users that are following this user.
 
+        .. warning::
+
+            The endpoint this method uses has been deprecated by Twitch and is no longer available.
+
         Parameters
         -----------
         token: Optional[:class:`str`]
@@ -524,11 +542,65 @@ async def fetch_followers(self, token: Optional[str] = None):
         data = await self._http.get_user_follows(token=token, to_id=str(self.id))
         return [FollowEvent(self._http, d, to=self) for d in data]
 
+    async def fetch_channel_followers(self, token: str, user_id: Optional[int] = None) -> List[ChannelFollowerEvent]:
+        """|coro|
+
+        Fetches a list of users that are following this broadcaster.
+        Requires a user access token that includes the moderator:read:followers scope.
+        The ID in the broadcaster_id query parameter must match the user ID in the access token or the user must be a moderator for the specified broadcaster.
+
+        Parameters
+        -----------
+        token: :class:`str`
+            User access token that includes the ``moderator:read:followers`` scope for the channel.
+        user_id: Optional[:class:`int`]
+            Use this parameter to see whether the user follows this broadcaster.
+
+
+        Returns
+        --------
+        List[:class:`twitchio.FollowEvent`]
+        """
+        from .models import ChannelFollowerEvent
+
+        data = await self._http.get_channel_followers(token=token, broadcaster_id=str(self.id), user_id=user_id)
+        return [ChannelFollowerEvent(self._http, d) for d in data]
+
+    async def fetch_channel_following(
+        self, token: str, broadcaster_id: Optional[int] = None
+    ) -> List[ChannelFollowingEvent]:
+        """|coro|
+
+        Fetches a list of users that this user follows.
+        Requires a user access token that includes the ``user:read:follows`` scope.
+        The ID in the broadcaster_id query parameter must match the user ID in the access token or the user must be a moderator for the specified broadcaster.
+
+        Parameters
+        -----------
+        token: :class:`str`
+            User access token that includes the moderator:read:followers scope for the channel.
+        broadcaster_id: Optional[:class:`int`]
+            Use this parameter to see whether the user follows this broadcaster.
+
+
+        Returns
+        --------
+        List[:class:`twitchio.ChannelFollowingEvent`]
+        """
+        from .models import ChannelFollowingEvent
+
+        data = await self._http.get_channel_followed(token=token, user_id=str(self.id), broadcaster_id=broadcaster_id)
+        return [ChannelFollowingEvent(self._http, d) for d in data]
+
     async def fetch_follow(self, to_user: "PartialUser", token: Optional[str] = None):
         """|coro|
 
         Check if a user follows another user or when they followed a user.
 
+        .. warning::
+
+            The endpoint this method uses has been deprecated by Twitch and is no longer available.
+
         Parameters
         -----------
         to_user: :class:`PartialUser`
@@ -546,10 +618,50 @@ async def fetch_follow(self, to_user: "PartialUser", token: Optional[str] = None
         data = await self._http.get_user_follows(token=token, from_id=str(self.id), to_id=str(to_user.id))
         return FollowEvent(self._http, data[0]) if data else None
 
+    async def fetch_channel_follower_count(self, token: Optional[str] = None) -> int:
+        """|coro|
+
+        Fetches the total number of users that are following this user.
+
+        Parameters
+        -----------
+        token: Optional[:class:`str`]
+            An oauth token to use instead of the bots token
+
+        Returns
+        --------
+        :class:`int`
+        """
+
+        data = await self._http.get_channel_follower_count(token=token, broadcaster_id=str(self.id))
+        return data["total"]
+
+    async def fetch_channel_following_count(self, token: str) -> int:
+        """|coro|
+
+        Fetches the total number of users that the user is following.
+
+        Parameters
+        -----------
+        token: :class:`str`
+            An oauth token to use instead of the bots token
+
+        Returns
+        --------
+        :class:`int`
+        """
+
+        data = await self._http.get_channel_followed_count(token=token, user_id=str(self.id))
+        return data["total"]
+
     async def fetch_follower_count(self, token: Optional[str] = None) -> int:
         """|coro|
 
-        Fetches a list of users that are following this user.
+        Fetches the total number of users that are following this user.
+
+        .. warning::
+
+            The endpoint this method uses has been deprecated by Twitch and is no longer available.
 
         Parameters
         -----------
@@ -569,6 +681,10 @@ async def fetch_following_count(self, token: Optional[str] = None) -> int:
 
         Fetches a list of users that this user is following.
 
+        .. warning::
+
+            The endpoint this method uses has been deprecated by Twitch and is no longer available.
+
         Parameters
         -----------
         token: Optional[:class:`str`]