Merge Eventsub and Helix Predictions

Author: chillymosh

twitchio/eventsub/payloads.py

@@ -100,7 +100,7 @@
     "CharityCampaignStopSubscription",
     "ChannelPredictionBeginSubscription",
     "ChannelPredictionLockSubscription",
-    "ChannelPredictionProgressubscription",
+    "ChannelPredictionProgressSubscription",
     "ChannelPredictionEndSubscription",
     "SharedChatSessionBeginSubscription",
     "SharedChatSessionEndSubscription",
@@ -1786,7 +1786,7 @@ def condition(self) -> Condition:
         return {"broadcaster_user_id": self.broadcaster_user_id}
 
 
-class ChannelPredictionProgressubscription(SubscriptionPayload):
+class ChannelPredictionProgressSubscription(SubscriptionPayload):
     """The ``channel.prediction.progress`` subscription type sends a notification when users participate in a Prediction on the specified channel.
 
     .. important::

twitchio/models/eventsub_.py

@@ -31,6 +31,7 @@
 from twitchio.models.channel_points import CustomReward
 from twitchio.models.charity import CharityValues
 from twitchio.models.chat import EmoteSet
+from twitchio.models.predictions import PredictionOutcome
 from twitchio.user import PartialUser
 from twitchio.utils import Colour, parse_timestamp
 
@@ -3143,6 +3144,56 @@ def __repr__(self) -> str:
         return f"<ChannelPollEnd broadcaster={self.broadcaster} id={self.id} title={self.title} started_at={self.started_at} ended_at={self.ended_at}>"
 
 
+class BaseChannelPrediction(BaseEvent):
+    def __init__(
+        self,
+        payload: ChannelPredictionBeginEvent
+        | ChannelPredictionProgressEvent
+        | ChannelPredictionLockEvent
+        | ChannelPredictionEndEvent,
+        *,
+        http: HTTPClient,
+    ) -> None:
+        self.id: str = payload["id"]
+        self.broadcaster: PartialUser = PartialUser(
+            payload["broadcaster_user_id"], payload["broadcaster_user_login"], http=http
+        )
+        self.title: str = payload["title"]
+        self.outcomes: list[PredictionOutcome] = [PredictionOutcome(c, http=http) for c in payload["outcomes"]]
+        self.started_at: datetime.datetime = parse_timestamp(payload["started_at"])
+
+    def __repr__(self) -> str:
+        return f"<BaseChannelPrediction id={self.id} title={self.title} started_at={self.started_at}>"
+
+
+class ChannelPredictionBegin(BaseChannelPrediction):
+    subscription_type = "channel.prediction.begin"
+
+    def __init__(self, data: ChannelPredictionBeginEvent, *, http: HTTPClient) -> None:
+        super().__init__(data, http=http)
+
+
+class ChannelPredictionProgress(BaseChannelPrediction):
+    subscription_type = "channel.prediction.progress"
+
+    def __init__(self, data: ChannelPredictionProgressEvent, *, http: HTTPClient) -> None:
+        super().__init__(data, http=http)
+
+
+class ChannelPredictionLock(BaseChannelPrediction):
+    subscription_type = "channel.prediction.lock"
+
+    def __init__(self, data: ChannelPredictionLockEvent, *, http: HTTPClient) -> None:
+        super().__init__(data, http=http)
+
+
+class ChannelPredictionEnd(BaseChannelPrediction):
+    subscription_type = "channel.prediction.end"
+
+    def __init__(self, data: ChannelPredictionEndEvent, *, http: HTTPClient) -> None:
+        super().__init__(data, http=http)
+
+
 class SuspiciousUserUpdate(BaseEvent):
     """
     Represents a suspicious user update event.

twitchio/models/predictions.py

@@ -34,6 +34,10 @@
     import datetime
 
     from twitchio.http import HTTPClient
+    from twitchio.types_.eventsub import (
+        PredictionOutcomeData,
+        PredictorData,
+    )
     from twitchio.types_.responses import (
         PredictionsResponseData,
         PredictionsResponseOutcomes,
@@ -44,16 +48,20 @@
 
 
 class Prediction:
-    """
-    Represents a Prediction
-
-    | Status      | Description |
-    | ----------- | ----------- |
-    | ACTIVE      | The Prediction is running and viewers can make predictions.   |
-    | CANCELED    | The broadcaster canceled the Prediction and refunded the Channel Points to the participants.  |
-    | LOCKED      | The broadcaster locked the Prediction, which means viewers can no longer make predictions.    |
-    | RESOLVED    | The winning outcome was determined and the Channel Points were distributed to the viewers who predicted the correct outcome.   |
-
+    """Represents a Prediction
+
+    +-----------+----------------------------------------------------------------------------------------------------------------+
+    | Status    | Description                                                                                                    |
+    +===========+================================================================================================================+
+    | ACTIVE    | The Prediction is running and viewers can make predictions.                                                    |
+    +-----------+----------------------------------------------------------------------------------------------------------------+
+    | CANCELED  | The broadcaster canceled the Prediction and refunded the Channel Points to the participants.                   |
+    +-----------+----------------------------------------------------------------------------------------------------------------+
+    | LOCKED    | The broadcaster locked the Prediction, which means viewers can no longer make predictions.                     |
+    +-----------+----------------------------------------------------------------------------------------------------------------+
+    | RESOLVED  | The winning outcome was determined and the Channel Points were distributed to the viewers who predicted the    |
+    |           | correct outcome.                                                                                               |
+    +-----------+----------------------------------------------------------------------------------------------------------------+
 
     Attributes
     ----------
@@ -72,9 +80,9 @@ class Prediction:
     created_at: datetime.datetime
         The datetime of when the prediction began.
     ended_at: datetime.datetime | None
-        The datetime of when the prediction ended. This is None if status is `ACTIVE`.
+        The datetime of when the prediction ended. This is None if status is ``ACTIVE``.
     locked_at: datetime.datetime | None
-        The datetime of when the prediction locked. If status is not `LOCKED`, this is set to None.
+        The datetime of when the prediction locked. If status is not ``LOCKED``, this is set to ``None``.
     """
 
     __slots__ = (
@@ -91,7 +99,12 @@ class Prediction:
         "locked_at",
     )
 
-    def __init__(self, data: PredictionsResponseData, *, http: HTTPClient) -> None:
+    def __init__(
+        self,
+        data: PredictionsResponseData,
+        *,
+        http: HTTPClient,
+    ) -> None:
         self._http = http
         self.id: str = data["id"]
         self.broadcaster: PartialUser = PartialUser(data["broadcaster_id"], data["broadcaster_login"], http=http)
@@ -116,12 +129,11 @@ async def end_prediction(
         token_for: str,
         winning_outcome_id: str | None = None,
     ) -> Prediction:
-        """
-        End an active prediction.
+        """End an active prediction.
 
         Parameters
         ----------
-        status  Literal["RESOLVED", "CANCELED", "LOCKED"]
+        status: Literal["RESOLVED", "CANCELED", "LOCKED"]
             The status to set the prediction to. Possible case-sensitive values are: `RESOLVED`, `CANCELED`, `LOCKED`
         token_for: str
             User access token that includes the `channel:manage:predictions` scope.
@@ -142,65 +154,69 @@ async def end_prediction(
 
 
 class PredictionOutcome:
-    """
-    Represents a prediction outcome.
+    """Represents a prediction outcome.
 
     Attributes
     ----------
     id: str
         An ID that identifies the choice.
     title: str
         The choice's title.
-    users: int
+    users: int | None
         The number of unique viewers that chose this outcome.
-    channel_points: int
+    channel_points: int | None
         The number of Channel Points spent by viewers on this outcome.
-    top_predictors: int
+    top_predictors: list[Predictor]
         A list of viewers who were the top predictors; otherwise, None if none.
-    colour: Literal["BLUE", "PINK"]
+    colour: Literal["BLUE", "PINK", "blue", "pink"]
         The number of votes cast using Channel Points.
     """
 
     __slots__ = ("id", "title", "colour", "channel_points", "users", "top_predictors")
 
-    def __init__(self, data: PredictionsResponseOutcomes, *, http: HTTPClient) -> None:
+    def __init__(
+        self,
+        data: PredictionsResponseOutcomes | PredictionOutcomeData,
+        *,
+        http: HTTPClient,
+    ) -> None:
         self.id: str = data["id"]
         self.title: str = data["title"]
-        self.users: int = int(data["users"])
-        self.channel_points: int = int(data["channel_points"])
-        self.colour: str = data["color"]
-        self.top_predictors: list[Predictor] | None = (
-            [Predictor(d, http=http) for d in data["top_predictors"]] if data["top_predictors"] else None
-        )
+        self.colour: Literal["BLUE", "PINK", "blue", "pink"] = data["color"]
+        users = data.get("users")
+        self.users: int | None = int(users) if users is not None else None
+        channel_points = data.get("channel_points")
+        self.channel_points: int | None = int(channel_points) if channel_points is not None else None
+        self.top_predictors: list[Predictor] = [Predictor(d, http=http) for d in data.get("top_predictors", [])]
 
     @property
     def color(self) -> str:
         """The color of the prediction. Alias to colour."""
-        return self.color
+        return self.colour
 
     def __repr__(self) -> str:
         return f"<PredictionOutcome id={self.id} title={self.title} channel_points={self.channel_points}>"
 
 
 class Predictor:
-    """
-    Represents a predictor
+    """Represents a predictor
 
     Attributes
     -----------
     user: PartialUser
         The viewer.
     channel_points_used: int
         Number of Channel Points used by the user.
-    channel_points_won: int
+    channel_points_won: int | None
         Number of Channel Points won by the user.
     """
 
     __slots__ = ("channel_points_used", "channel_points_won", "user")
 
-    def __init__(self, data: PredictionsResponseTopPredictors, *, http: HTTPClient) -> None:
+    def __init__(self, data: PredictionsResponseTopPredictors | PredictorData, *, http: HTTPClient) -> None:
         self.channel_points_used: int = data["channel_points_used"]
-        self.channel_points_won: int = data["channel_points_won"]
+        channel_points_won = data.get("channel_points_won")
+        self.channel_points_won: int | None = int(channel_points_won) if channel_points_won is not None else None
         self.user = PartialUser(data["user_id"], data["user_login"], http=http)
 
     def __repr__(self) -> str:

twitchio/types_/eventsub.py

@@ -999,16 +999,16 @@ class PredictorData(BaseUserEvent):
 class PredictionOutcomeData(TypedDict):
     id: str
     title: str
-    color: str
-    users: int
+    color: Literal["blue", "pink"]
+    users: NotRequired[int]
     channel_points: NotRequired[int]
     top_predictors: NotRequired[list[PredictorData]]
 
 
 class BaseChannelPredictionData(BaseBroadcasterEvent):
     id: str
     title: str
-    outcomes: list[PredictorData]
+    outcomes: list[PredictionOutcomeData]
     started_at: str