Add Video methods models

twitchio/client.py

@@ -43,6 +43,7 @@
     SearchChannel,
     Stream,
     Team,
+    Video,
 )
 from .payloads import EventErrorPayload
 from .web import AiohttpAdapter
@@ -132,7 +133,8 @@ def dispatch(self, event: str, payload: Any | None = None) -> None:
 
         _ = [asyncio.create_task(self._dispatch(listener, original=payload)) for listener in listeners]
 
-    async def setup_hook(self) -> None: ...
+    async def setup_hook(self) -> None:
+        ...
 
     async def login(self, *, token: str | None = None) -> None:
         if not token:
@@ -335,8 +337,11 @@ async def fetch_clips(
         :class:`~twitchio.HTTPAsyncIterator`[:class:`~twitchio.Clip`]
         """
 
-        if sum(x is not None for x in [broadcaster_id, game_id, clip_ids]) > 1:
-            raise ValueError("The parameters 'broadcaster_id', 'game_id', and 'ids' are mutually exclusive.")
+        provided: int = len([v for v in (broadcaster_id, game_id, clip_ids) if v])
+        if provided > 1:
+            raise ValueError("Only one of 'name', 'id', or 'igdb_id' can be provided.")
+        elif provided == 0:
+            raise ValueError("One of 'name', 'id', or 'igdb_id' must be provided.")
 
         first = max(1, min(100, first))
 
@@ -617,6 +622,101 @@ async def search_channels(
             token_for=token_for,
         )
 
+    async def fetch_videos(
+        self,
+        *,
+        ids: list[str | int] | None = None,
+        user_id: str | int | None = None,
+        game_id: str | int | None = None,
+        language: str | None = None,
+        period: Literal["all", "day", "month", "week"] = "all",
+        sort: Literal["time", "trending", "views"] = "time",
+        type: Literal["all", "archive", "highlight", "upload"] = "all",
+        first: int = 20,
+        token_for: str | None = None,
+    ) -> HTTPAsyncIterator[Video]:
+        """Fetch a list of [`twitchio.Video`][twitchio.Video] objects with the provided `ids`, `user_id` or `game_id`.
+
+        One of `ids`, `user_id` or `game_id` must be provided.
+        If more than one is provided or no parameters are provided, a `ValueError` will be raised.
+
+
+        Parameters
+        ----------
+        ids: list[str | int] | None
+            A list of video IDs to fetch.
+        user_id: str | int | None
+            The ID of the user whose list of videos you want to get.
+        game_id: str | int | None
+            The igdb_id of the game to fetch.
+        language: str | None
+        period: Literal["all", "day", "month", "week"]
+        sort: Literal["time", "trending", "views"]
+        type: Literal["all", "archive", "highlight", "upload"]
+        first: int
+        token_for: str | None
+            An optional User OAuth token to use instead of the default app token.
+
+        Returns
+        -------
+        list[Video]
+            A list of Video objects if found.
+
+        Raises
+        ------
+        ValueError
+            Only one of the 'ids', 'user_id', or 'game_id' parameters can be provided.
+        ValueError
+            One of the 'ids', 'user_id', or 'game_id' parameters must be provided.
+        """
+        provided: int = len([v for v in (ids, game_id, user_id) if v])
+        if provided > 1:
+            raise ValueError("Only one of 'ids', 'user_id', or 'game_id' can be provided.")
+        elif provided == 0:
+            raise ValueError("One of 'name', 'id', or 'igdb_id' must be provided.")
+
+        first = max(1, min(100, first))
+
+        return await self._http.get_videos(
+            ids=ids,
+            user_id=user_id,
+            game_id=game_id,
+            language=language,
+            period=period,
+            sort=sort,
+            type=type,
+            first=first,
+            token_for=token_for,
+        )
+
+    async def delete_videos(self, ids: list[str | int], token_for: str) -> list[str]:
+        """Deletes one or more videos. You may delete past broadcasts, highlights, or uploads.
+
+        This requires a user token with the scope ``channel:manage:videos``.
+        The limit is to delete 5 ids at a time, so if more than 5 ids are provided we will attempt to delete them in chunks.
+        If any of the videos fail to delete in the request then none will be deleted in that chunk.
+
+        Parameters
+        ----------
+        ids: list[str | int] | None
+            A list of video IDs to fetch.
+        token_for: str
+            A User OAuth token with the scope ``channel:manage:videos``.
+
+        Returns
+        -------
+        list[str]
+            A list of Video IDs that were successfully deleted.
+        """
+        resp: list[str] = []
+
+        for chunk in [ids[x : x + 5] for x in range(0, len(ids), 5)]:
+            data = await self._http.delete_videos(ids=chunk, token_for=token_for)
+            if data:
+                resp.extend(data["data"])
+
+        return resp
+
     def doc_test(self, thing: int = 1) -> int:
         """This is a test method to test and view certain elements of the mkdocs style documentation.
 

twitchio/http.py

@@ -36,7 +36,7 @@
 
 from . import __version__
 from .exceptions import HTTPException
-from .models import Clip, Game, SearchChannel, Stream
+from .models import Clip, Game, SearchChannel, Stream, Video
 from .utils import _from_json  # type: ignore
 
 
@@ -57,6 +57,8 @@
         SearchChannelResponse,
         StreamResponse,
         TeamPayload,
+        VideoDeletePayload,
+        VideoResponse,
     )
 
 
@@ -557,3 +559,39 @@ async def converter(data: GameResponse) -> Game:
 
         iterator: HTTPAsyncIterator[Game] = self.request_paginated(route, converter=converter)
         return iterator
+
+    async def get_videos(
+        self,
+        ids: list[str | int] | None = None,
+        user_id: str | int | None = None,
+        game_id: str | int | None = None,
+        language: str | None = None,
+        period: Literal["all", "day", "month", "week"] = "all",
+        sort: Literal["time", "trending", "views"] = "time",
+        type: Literal["all", "archive", "highlight", "upload"] = "all",
+        first: int = 20,
+        token_for: str | None = None,
+    ) -> HTTPAsyncIterator[Video]:
+        params: dict[str, int | str | list[str | int]] = {"first": first, "period": period, "sort": sort, "type": type}
+
+        if ids is not None:
+            params["id"] = ids
+        if user_id is not None:
+            params["user_id"] = user_id
+        if game_id is not None:
+            params["game_id"] = game_id
+        if language is not None:
+            params["language"] = language
+
+        route = Route("GET", "videos", params=params, token_for=token_for)
+
+        async def converter(data: VideoResponse) -> Video:
+            return Video(data, http=self)
+
+        iterator = self.request_paginated(route, converter=converter)
+        return iterator
+
+    async def delete_videos(self, ids: list[str | int], token_for: str) -> VideoDeletePayload:
+        params = {"id": ids}
+        route: Route = Route("DELETE", "videos", params=params, token_for=token_for)
+        return await self.request_json(route)

twitchio/models.py

@@ -48,6 +48,7 @@
         SearchChannelResponse,
         StreamResponse,
         TeamResponse,
+        VideoResponse,
     )
 
 
@@ -63,6 +64,7 @@
     "SearchChannel",
     "Stream",
     "Team",
+    "Video",
 )
 
 
@@ -657,3 +659,88 @@ def __eq__(self, __value: object) -> bool:
             return NotImplemented
 
         return __value.id == self.id
+
+
+class Video:
+    """
+    Represents video information
+
+    Attributes
+    -----------
+    id: :class:`int`
+        The ID of the video.
+    user: :class:`~twitchio.PartialUser`
+        User who owns the video.
+    title: :class:`str`
+        Title of the video
+    description: :class:`str`
+        Description of the video.
+    created_at: :class:`datetime.datetime`
+        Date when the video was created.
+    published_at: :class:`datetime.datetime`
+       Date when the video was published.
+    url: :class:`str`
+        URL of the video.
+    thumbnail_url: :class:`str`
+        Template URL for the thumbnail of the video.
+    viewable: :class:`str`
+        Indicates whether the video is public or private.
+    view_count: :class:`int`
+        Number of times the video has been viewed.
+    language: :class:`str`
+        Language of the video.
+    type: :class:`str`
+        The type of video.
+    duration: :class:`str`
+        Length of the video.
+    """
+
+    __slots__ = (
+        "_http",
+        "id",
+        "user",
+        "title",
+        "description",
+        "created_at",
+        "published_at",
+        "url",
+        "thumbnail_url",
+        "viewable",
+        "view_count",
+        "language",
+        "type",
+        "duration",
+    )
+
+    def __init__(self, data: VideoResponse, *, http: HTTPClient) -> None:
+        self._http: HTTPClient = http
+        self.id: str = data["id"]
+        self.user = PartialUser(data["user_id"], data["user_name"])
+        self.title: str = data["title"]
+        self.description: str = data["description"]
+        self.created_at = parse_timestamp(data["created_at"])
+        self.published_at = parse_timestamp(data["published_at"])
+        self.url: str = data["url"]
+        self.thumbnail_url: str = data["thumbnail_url"]
+        self.viewable: str = data["viewable"]
+        self.view_count: int = data["view_count"]
+        self.language: str = data["language"]
+        self.type: str = data["type"]
+        self.duration: str = data["duration"]
+
+    def __repr__(self) -> str:
+        return f"<Video id={self.id} title={self.title} url={self.url}>"
+
+    async def delete(self, token_for: str) -> None:
+        """|coro|
+
+        Deletes the video. For bulk deletion see :func:`Client.delete_videos`
+
+        Parameters
+        -----------
+        ids: list[str | int]
+            List of video IDs to delete
+        token_for: str
+            A user oauth token with the channel:manage:videos
+        """
+        await self._http.delete_videos(ids=[self.id], token_for=token_for)

twitchio/types_/responses.py

@@ -58,6 +58,9 @@
     "TeamMemberResponse",
     "TeamResponse",
     "TeamPayload",
+    "VideoDeletePayload",
+    "VideoResponse",
+    "VideoPayload",
 )
 
 T = TypeVar("T")
@@ -113,6 +116,10 @@ class AuthorizationURLResponse(TypedDict):
 RawResponse: TypeAlias = dict[str, Any]
 
 
+class Pagination(TypedDict):
+    cursor: str | None
+
+
 class ChatterColorResponse(TypedDict):
     user_id: str
     user_login: str
@@ -245,6 +252,43 @@ class TeamResponse(TypedDict):
     id: str
 
 
+class MutedSegment(TypedDict):
+    duration: int
+    offset: int
+
+
+class VideoResponse(TypedDict):
+    id: str
+    stream_id: str | None
+    user_id: str
+    user_login: str
+    user_name: str
+    title: str
+    description: str
+    created_at: str
+    published_at: str
+    url: str
+    thumbnail_url: str
+    viewable: str
+    view_count: int
+    language: str
+    type: str
+    duration: str
+    muted_segments: list[MutedSegment] | None
+
+
+class StreamPayload(TypedDict):
+    data: list[StreamResponse]
+    pagination: Pagination
+
+
+class VideoPayload(TypedDict):
+    data: list[VideoResponse]
+    pagination: Pagination
+
+class VideoDeletePayload(TypedDict):
+    data: list[str]
+
 ChatterColorPayload = Payload[ChatterColorResponse]
 ChannelInfoPayload = Payload[ChannelInfoResponse]
 ClipPayload = Payload[ClipResponse]
@@ -253,5 +297,4 @@ class TeamResponse(TypedDict):
 GamePayload = Payload[GameResponse]
 GlobalEmotePayload = Payload[GlobalEmoteResponse]
 SearchChannelPayload = Payload[SearchChannelResponse]
-StreamPayload = Payload[StreamResponse]
 TeamPayload = Payload[TeamResponse]