Add Poll endpoints and tidied schedule

Added poll endpoints with documentation

Moved validation checks for schedule into user rather than http.

Author: chillymosh

docs/changelog.rst

@@ -8,6 +8,7 @@ Massive documentation updates
     - Removed unexpected loop termination from `WSConnection._close()`
     - Fix bug where # prefixed channel names in initial_channels would not trigger :func:`twitchio.Client.event_ready`
     - :func:`User.create_clip` has been fixed by converting bool to string in http request
+    - Poll endpoints added :func:`User.fetch_polls` :func:`User.create_poll` and :func:`User.end_poll`
 
 - ext.commands
     - :func:`Bot.handle_commands` now also invokes on threads / replies
@@ -30,7 +31,7 @@ Massive documentation updates
     - Fix :func:`twitchio.Client.wait_for_ready`
     - Remove loop= parameter inside :func:`twitchio.Client.wait_for` for 3.10 compatibility
     - Add ``is_broadcaster`` check to :class:`twitchio.PartialChatter`. This is accessible as ``Context.author.is_broadcaster``
-    - :func:`twitchio.User.fetch_follow` will now return ``None`` if the FollowEvent does not exists
+    - :func:`User.fetch_follow` will now return ``None`` if the FollowEvent does not exists
     - TwitchIO will now correctly handle error raised when only the prefix is typed in chat
     - Fix paginate logic in :func:`TwitchHTTP.request`
 

docs/reference.rst

@@ -207,6 +207,20 @@ ModEvent
     :members:
     :inherited-members:
 
+Poll
+-------------
+.. attributetable:: Poll
+
+.. autoclass:: Poll
+    :members:
+    :inherited-members:
+
+.. attributetable:: PollChoice
+
+.. autoclass:: PollChoice
+    :members:
+    :inherited-members:
+
 Predictions
 -------------
 .. attributetable:: Prediction

twitchio/http.py

@@ -220,7 +220,7 @@ async def _request(self, route, path, headers, utilize_bucket=True):
 
                 if 500 <= resp.status <= 504:
                     reason = resp.reason
-                    await asyncio.sleep(2**attempt + 1)
+                    await asyncio.sleep(2 ** attempt + 1)
                     continue
 
                 if utilize_bucket:
@@ -251,7 +251,7 @@ async def _request(self, route, path, headers, utilize_bucket=True):
                     reason = "Ratelimit Reached"
 
                     if not utilize_bucket:  # non Helix APIs don't have ratelimit headers
-                        await asyncio.sleep(3**attempt + 1)
+                        await asyncio.sleep(3 ** attempt + 1)
                     continue
 
                 raise errors.HTTPException(
@@ -584,7 +584,9 @@ async def post_prediction(
 
     async def post_create_clip(self, token: str, broadcaster_id: int, has_delay=False):
         return await self.request(
-            Route("POST", "clips", query=[("broadcaster_id", broadcaster_id), ("has_delay", str(has_delay))], token=token),
+            Route(
+                "POST", "clips", query=[("broadcaster_id", broadcaster_id), ("has_delay", str(has_delay))], token=token
+            ),
             paginate=False,
         )
 
@@ -759,23 +761,11 @@ async def patch_channel(
     async def get_channel_schedule(
         self,
         broadcaster_id: str,
-        segment_ids: List[str] = None,
-        start_time: datetime.datetime = None,
-        utc_offset: int = None,
+        segment_ids: Optional[List[str]] = None,
+        start_time: Optional[datetime.datetime] = None,
+        utc_offset: Optional[int] = None,
         first: int = 20,
     ):
-
-        if first is not None and (first > 25 or first < 1):
-            raise ValueError("The parameter 'first' was malformed: the value must be less than or equal to 25")
-        if segment_ids is not None and len(segment_ids) > 100:
-            raise ValueError("segment_id can only have 100 entries")
-
-        if start_time:
-            start_time = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
-
-        if utc_offset:
-            utc_offset = str(utc_offset)
-
         q = [
             x
             for x in [
@@ -927,3 +917,46 @@ async def get_teams(self, team_name: str = None, team_id: str = None):
     async def get_channel_teams(self, broadcaster_id: str):
         q = [("broadcaster_id", broadcaster_id)]
         return await self.request(Route("GET", "teams/channel", query=q))
+
+    async def get_polls(
+        self,
+        broadcaster_id: str,
+        token: str,
+        poll_ids: Optional[List[str]] = None,
+        first: Optional[int] = 20,
+    ):
+        q = [("broadcaster_id", broadcaster_id), ("first", first)]
+        if poll_ids:
+            q.extend(("id", poll_id) for poll_id in poll_ids)
+        return await self.request(Route("GET", "polls", query=q, token=token), paginate=False, full_body=True)
+
+    async def post_poll(
+        self,
+        broadcaster_id: str,
+        token: str,
+        title: str,
+        choices,
+        duration: int,
+        bits_voting_enabled: Optional[bool] = False,
+        bits_per_vote: Optional[int] = None,
+        channel_points_voting_enabled: Optional[bool] = False,
+        channel_points_per_vote: Optional[int] = None,
+    ):
+        body = {
+            "broadcaster_id": broadcaster_id,
+            "title": title,
+            "choices": [{"title": choice} for choice in choices],
+            "duration": duration,
+            "bits_voting_enabled": str(bits_voting_enabled),
+            "channel_points_voting_enabled": str(channel_points_voting_enabled),
+        }
+        if bits_voting_enabled and bits_per_vote:
+            body["bits_per_vote"] = bits_per_vote
+        if channel_points_voting_enabled and channel_points_per_vote:
+            body["channel_points_per_vote"] = channel_points_per_vote
+
+        return await self.request(Route("POST", "polls", body=body, token=token))
+
+    async def patch_poll(self, broadcaster_id: str, token: str, id: str, status: str):
+        body = {"broadcaster_id": broadcaster_id, "id": id, "status": status}
+        return await self.request(Route("PATCH", "polls", body=body, token=token))

twitchio/models.py

@@ -68,6 +68,8 @@
     "Team",
     "ChannelTeams",
     "ChannelInfo",
+    "Poll",
+    "PollChoice",
 )
 
 
@@ -1344,3 +1346,98 @@ def __init__(self, http: "TwitchHTTP", data: dict):
 
     def __repr__(self):
         return f"<ChannelTeams user={self.broadcaster} team_name={self.team_name} team_display_name={self.team_display_name} id={self.id} created_at={self.created_at}>"
+
+
+class Poll:
+    """
+    Represents a list of Polls for a broadcaster / channel
+
+    Attributes
+    -----------
+    id: :class:`str`
+        ID of a poll.
+    broadcaster: :class:`~twitchio.PartialUser`
+        User of the broadcaster.
+    title: :class:`str`
+        Question displayed for the poll.
+    choices: List[:class:`~twitchio.PollChoice`]
+        The poll choices.
+    bits_voting_enabled: :class:`bool`
+        Indicates if Bits can be used for voting.
+    bits_per_vote: :class:`int`
+        Number of Bits required to vote once with Bits.
+    channel_points_voting_enabled: :class:`bool`
+        Indicates if Channel Points can be used for voting.
+    channel_points_per_vote: :class:`int`
+        Number of Channel Points required to vote once with Channel Points.
+    status: :class:`str`
+        Poll status. Valid values: ACTIVE, COMPLETED, TERMINATED, ARCHIVED, MODERATED, INVALID
+    duration: :class:`int`
+        Total duration for the poll (in seconds).
+    started_at: :class:`datetime.datetime`
+        Date and time the the poll was started.
+    ended_at: :class:`datetime.datetime`
+        Date and time the the poll was ended.
+    """
+
+    __slots__ = (
+        "id",
+        "broadcaster",
+        "title",
+        "choices",
+        "channel_points_voting_enabled",
+        "channel_points_per_vote",
+        "status",
+        "duration",
+        "started_at",
+        "ended_at",
+    )
+
+    def __init__(self, http: "TwitchHTTP", data: dict):
+        self.id: str = data["id"]
+        self.broadcaster = PartialUser(http, data["broadcaster_id"], data["broadcaster_login"])
+        self.title: str = data["title"]
+        self.choices: List[PollChoice] = [PollChoice(d) for d in data["choices"]] if data["choices"] else []
+        self.channel_points_voting_enabled: bool = data["channel_points_voting_enabled"]
+        self.channel_points_per_vote: int = data["channel_points_per_vote"]
+        self.status: str = data["status"]
+        self.duration: int = data["duration"]
+        self.started_at: datetime.datetime = parse_timestamp(data["started_at"])
+        try:
+            self.ended_at: Optional[datetime.datetime] = parse_timestamp(data["ended_at"])
+        except KeyError:
+            self.ended_at = None
+
+    def __repr__(self):
+        return f"<Polls id={self.id} broadcaster={self.broadcaster} title={self.title} status={self.status} duration={self.duration} started_at={self.started_at} ended_at={self.ended_at}>"
+
+
+class PollChoice:
+    """
+    Represents a polls choices
+
+    Attributes
+    -----------
+    id: :class:`str`
+        ID for the choice.
+    title: :class:`str`
+        Text displayed for the choice.
+    votes: :class:`int`
+        Total number of votes received for the choice across all methods of voting.
+    channel_points_votes: :class:`int`
+        Number of votes received via Channel Points.
+    bits_votes: :class:`int`
+        Number of votes received via Bits.
+    """
+
+    __slots__ = ("id", "title", "votes", "channel_points_votes", "bits_votes")
+
+    def __init__(self, data: dict):
+        self.id: str = data["id"]
+        self.title: str = data["title"]
+        self.votes: int = data["votes"]
+        self.channel_points_votes: int = data["channel_points_votes"]
+        self.bits_votes: int = data["bits_votes"]
+
+    def __repr__(self):
+        return f"<PollChoice id={self.id} title={self.title} votes={self.votes} channel_points_votes={self.channel_points_votes} bits_votes={self.bits_votes}>"