Add video methods (#8)

Author: joostlek

.editorconfig

@@ -7,6 +7,7 @@ indent_style = space
 insert_final_newline = true
 trim_trailing_whitespace = true
 indent_size = 4
+max_line_length = 88
 
 [*.md]
 trim_trailing_whitespace = false

pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
   "Topic :: Software Development :: Libraries :: Python Modules",
 ]
 packages = [
-    { include = "python_youtube", from = "src" },
+    { include = "async_python_youtube", from = "src" },
 ]
 
 [tool.poetry.dependencies]
@@ -57,7 +57,7 @@ show_missing = true
 
 [tool.coverage.run]
 plugins = ["covdefaults"]
-source = ["python_youtube"]
+source = ["async_python_youtube"]
 
 [tool.mypy]
 # Specify the target platform details in config, so your developers are
@@ -120,6 +120,7 @@ disable = [
   "duplicate-code",
   "format",
   "unsubscriptable-object",
+  "unnecessary-dunder-call",
 ]
 
 [tool.pylint.SIMILARITIES]
@@ -150,7 +151,7 @@ fixture-parentheses = false
 mark-parentheses = false
 
 [tool.ruff.isort]
-known-first-party = ["python_youtube"]
+known-first-party = ["async_python_youtube"]
 
 [tool.ruff.mccabe]
 max-complexity = 25

src/python_youtube/__init__.py src/async_python_youtube/__init__.pysrc/python_youtube/__init__.py renamed to src/async_python_youtube/__init__.py

@@ -0,0 +1,34 @@
+"""Models for YouTube API."""
+from enum import Enum
+
+
+class HttpStatusCode(int, Enum):
+    """Enum holding http status codes."""
+
+    NOT_FOUND = 404
+
+
+class VideoPart(str, Enum):
+    """Enum holding the part parameters for video requests."""
+
+    CONTENT_DETAILS = "contentDetails"
+    FILE_DETAILS = "fileDetails"
+    ID = "id"
+    LIVE_STREAMING_DETAILS = "liveStreamingDetails"
+    LOCALIZATIONS = "localizations"
+    PLAYER = "player"
+    PROCESSING_DETAILS = "processingDetails"
+    RECORDING_DETAILS = "recordingDetails"
+    SNIPPET = "snippet"
+    STATISTICS = "statistics"
+    STATUS = "status"
+    SUGGESTIONS = "suggestions"
+    TOPIC_DETAILS = "topicDetails"
+
+
+class LiveBroadcastContent(str, Enum):
+    """Enum holding the liveBroadcastContent values."""
+
+    NONE = "none"
+    LIVE = "live"
+    UPCOMING = "upcoming"

src/async_python_youtube/const.py

@@ -0,0 +1,21 @@
+"""Asynchronous Python client for the YouTube API."""
+
+
+class YouTubeError(Exception):
+    """Generic exception."""
+
+
+class YouTubeConnectionError(YouTubeError):
+    """YouTube connection exception."""
+
+
+class YouTubeCoordinateError(YouTubeError):
+    """YouTube coordinate exception."""
+
+
+class YouTubeUnauthenticatedError(YouTubeError):
+    """YouTube unauthenticated exception."""
+
+
+class YouTubeNotFoundError(YouTubeError):
+    """YouTube not found exception."""

src/async_python_youtube/exceptions.py

@@ -0,0 +1,35 @@
+"""Helper functions for the YouTube API."""
+from collections.abc import AsyncGenerator, Generator
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+async def first(generator: AsyncGenerator[T, None]) -> T | None:
+    """Return the first value or None from the given AsyncGenerator."""
+    try:
+        return await generator.__anext__()
+    except StopAsyncIteration:
+        return None
+
+
+def chunk(source: list[T], chunk_size: int) -> Generator[list[T], None, None]:
+    """Divide the source list in chunks of given size."""
+    for i in range(0, len(source), chunk_size):
+        yield source[i : i + chunk_size]
+
+
+async def limit(
+    generator: AsyncGenerator[T, None],
+    total: int,
+) -> AsyncGenerator[T, None]:
+    """Limit the number of entries returned from the AsyncGenerator."""
+    if total < 1:
+        msg = "Limit has to be an int > 1"
+        raise ValueError(msg)
+    count = 0
+    async for item in generator:
+        count += 1
+        if count > total:
+            break
+        yield item

src/async_python_youtube/helper.py

@@ -0,0 +1,52 @@
+"""Models for YouTube API."""
+from datetime import datetime
+from typing import TypeVar
+
+from pydantic import BaseModel, Field
+
+from async_python_youtube.const import LiveBroadcastContent
+
+T = TypeVar("T")
+
+
+class YouTubeVideoThumbnail(BaseModel):
+    """Model representing a video thumbnail."""
+
+    url: str = Field(...)
+    width: int = Field(...)
+    height: int = Field(...)
+
+
+class YouTubeVideoThumbnails(BaseModel):
+    """Model representing video thumbnails."""
+
+    default: YouTubeVideoThumbnail = Field(...)
+    medium: YouTubeVideoThumbnail = Field(...)
+    high: YouTubeVideoThumbnail = Field(...)
+    standard: YouTubeVideoThumbnail = Field(...)
+    maxres: YouTubeVideoThumbnail | None = Field(None)
+
+
+class YouTubeVideoSnippet(BaseModel):
+    """Model representing video snippet."""
+
+    published_at: datetime = Field(..., alias="publishedAt")
+    channel_id: str = Field(..., alias="channelId")
+    title: str = Field(...)
+    description: str = Field(...)
+    thumbnails: YouTubeVideoThumbnails = Field(...)
+    channel_title: str = Field(..., alias="channelTitle")
+    tags: list[str] = Field(...)
+    live_broadcast_content: LiveBroadcastContent = Field(
+        ...,
+        alias="liveBroadcastContent",
+    )
+    default_language: str | None = Field(None, alias="defaultLanguage")
+    default_audio_language: str | None = Field(None, alias="defaultAudioLanguage")
+
+
+class YouTubeVideo(BaseModel):
+    """Model representing a video."""
+
+    video_id: str = Field(..., alias="id")
+    snippet: YouTubeVideoSnippet | None = None

src/async_python_youtube/models.py

@@ -0,0 +1,157 @@
+"""The YouTube API."""
+
+import asyncio
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
+from importlib import metadata
+from typing import Any, cast
+
+import async_timeout
+from aiohttp import ClientResponseError, ClientSession
+from aiohttp.hdrs import METH_GET
+from yarl import URL
+
+from async_python_youtube.const import HttpStatusCode
+from async_python_youtube.exceptions import (
+    YouTubeConnectionError,
+    YouTubeError,
+    YouTubeNotFoundError,
+)
+from async_python_youtube.helper import chunk, first
+from async_python_youtube.models import YouTubeVideo
+
+__all__ = [
+    "YouTube",
+]
+
+MAX_RESULTS_FOR_VIDEO = 50
+
+
+@dataclass
+class YouTube:
+    """YouTube API client."""
+
+    session: ClientSession | None = None
+    request_timeout: int = 10
+    api_host: str = "youtube.googleapis.com"
+    _close_session: bool = False
+
+    async def _request(
+        self,
+        uri: str,
+        *,
+        data: dict[str, Any] | None = None,
+        error_handler: dict[int, BaseException] | None = None,
+    ) -> dict[str, Any]:
+        """Handle a request to OpenSky.
+
+        A generic method for sending/handling HTTP requests done against
+        OpenSky.
+
+        Args:
+        ----
+            uri: the path to call.
+            data: the query parameters to add.
+
+        Returns:
+        -------
+            A Python dictionary (JSON decoded) with the response from
+            the API.
+
+        Raises:
+        ------
+            OpenSkyConnectionError: An error occurred while communicating with
+                the OpenSky API.
+            OpenSkyrror: Received an unexpected response from the OpenSky API.
+        """
+        version = metadata.version(__package__)
+        url = URL.build(
+            scheme="https",
+            host=self.api_host,
+            port=443,
+            path="/youtube/v3/",
+        ).joinpath(uri)
+
+        headers = {
+            "User-Agent": f"PythonOpenSky/{version}",
+            "Accept": "application/json, text/plain, */*",
+        }
+
+        if self.session is None:
+            self.session = ClientSession()
+            self._close_session = True
+
+        try:
+            async with async_timeout.timeout(self.request_timeout):
+                response = await self.session.request(
+                    METH_GET,
+                    url.with_query(data),
+                    headers=headers,
+                )
+                response.raise_for_status()
+        except asyncio.TimeoutError as exception:
+            msg = "Timeout occurred while connecting to the YouTube API"
+            raise YouTubeConnectionError(msg) from exception
+        except ClientResponseError as exception:
+            if error_handler and exception.status in error_handler:
+                raise error_handler[exception.status] from exception
+            msg = "Error occurred while communicating with YouTube API"
+            raise YouTubeConnectionError(msg) from exception
+
+        content_type = response.headers.get("Content-Type", "")
+
+        if "application/json" not in content_type:
+            text = await response.text()
+            msg = "Unexpected response from the YouTube API"
+            raise YouTubeError(
+                msg,
+                {"Content-Type": content_type, "response": text},
+            )
+
+        return cast(dict[str, Any], await response.json())
+
+    async def get_video(self, video_id: str) -> YouTubeVideo | None:
+        """Get a single video."""
+        return await first(self.get_videos([video_id]))
+
+    async def get_videos(
+        self,
+        video_ids: list[str],
+    ) -> AsyncGenerator[YouTubeVideo, None]:
+        """Get a list of videos."""
+        error_handler: dict[int, BaseException] = {
+            HttpStatusCode.NOT_FOUND: YouTubeNotFoundError("Video not found"),
+        }
+        for video_chunk in chunk(video_ids, MAX_RESULTS_FOR_VIDEO):
+            ids = ",".join(video_chunk)
+            data = {
+                "part": "snippet",
+                "id": ids,
+                "maxResults": MAX_RESULTS_FOR_VIDEO,
+            }
+            res = await self._request("videos", data=data, error_handler=error_handler)
+            for item in res["items"]:
+                yield YouTubeVideo.parse_obj(item)
+
+    async def close(self) -> None:
+        """Close open client session."""
+        if self.session and self._close_session:
+            await self.session.close()
+
+    async def __aenter__(self) -> Any:
+        """Async enter.
+
+        Returns
+        -------
+            The YouTube object.
+        """
+        return self
+
+    async def __aexit__(self, *_exc_info: Any) -> None:
+        """Async exit.
+
+        Args:
+        ----
+            _exc_info: Exec type.
+        """
+        await self.close()