Overhaul (#46)

Author: joostlek

pyproject.toml

@@ -121,6 +121,9 @@ disable = [
   "format",
   "unsubscriptable-object",
   "unnecessary-dunder-call",
+  "too-many-instance-attributes",
+  "too-many-arguments",
+  "too-many-locals",
 ]
 
 [tool.pylint.SIMILARITIES]
@@ -144,6 +147,9 @@ ignore = [
   "PLR2004", # Just annoying, not really useful
   "TCH001",
   "SLOT000",
+  "FBT001", # Boolean positional arg
+  "FBT002", # Boolean default arg
+  "PLR0913", # A lot of arguments
 ]
 select = ["ALL"]
 

src/async_python_youtube/exceptions.py

@@ -1,9 +1,82 @@
 """Helper functions for the YouTube API."""
+import urllib.parse
 from collections.abc import AsyncGenerator, Generator
-from typing import TypeVar
+from enum import Enum
+from typing import Any, TypeVar
+
+from async_python_youtube.types import AuthScope
+
+__all__ = [
+    "YOUTUBE_AUTH_BASE_URL",
+    "YOUTUBE_AUTH_TOKEN_URL",
+    "build_scope",
+    "build_url",
+    "first",
+    "chunk",
+    "limit",
+]
 
 T = TypeVar("T")
 
+YOUTUBE_AUTH_BASE_URL: str = "https://oauth2.googleapis.com"
+YOUTUBE_AUTH_TOKEN_URL: str = f"{YOUTUBE_AUTH_BASE_URL}/token"
+
+
+def build_scope(scopes: list[AuthScope]) -> str:
+    """Build a valid scope string from list.
+
+    :param scopes: list of :class:`~async_python_youtube.types.AuthScope`
+    :returns: the valid auth scope string
+    """
+    return " ".join([s.value for s in scopes])
+
+
+def build_url(
+    url: str,
+    params: dict[str, Any],
+    remove_none: bool = False,
+    split_lists: bool = False,
+    enum_value: bool = True,
+) -> str:
+    """Build a valid url string.
+
+    :param url: base URL
+    :param params: dictionary of URL parameter
+    :param remove_none: if set all params that have a None value get removed
+    |default| :code:`False`
+    :param split_lists: if set all params that are a list will be split over multiple
+    url parameter with the same name |default| :code:`False`
+    :param enum_value: if true, automatically get value string from Enum values
+    |default| :code:`True`
+    :return: URL
+    """
+
+    def get_value(input_value: Any) -> str:
+        if not enum_value:
+            return str(input_value)
+        if isinstance(input_value, Enum):
+            return str(input_value.value)
+        return str(input_value)
+
+    def add_param(res: str, query_key: str, query_value: Any) -> str:
+        if len(res) > 0:
+            res += "&"
+        res += query_key
+        if query_value is not None:
+            res += "=" + urllib.parse.quote(get_value(query_value))
+        return res
+
+    result = ""
+    for key, value in params.items():
+        if value is None and remove_none:
+            continue
+        if split_lists and isinstance(value, list):
+            for val in value:
+                result = add_param(result, key, val)
+        else:
+            result = add_param(result, key, value)
+    return url + (("?" + result) if len(result) > 0 else "")
+
 
 async def first(generator: AsyncGenerator[T, None]) -> T | None:
     """Return the first value or None from the given AsyncGenerator."""

src/async_python_youtube/helper.py

@@ -6,10 +6,22 @@
 
 from async_python_youtube.const import LiveBroadcastContent
 
+__all__ = [
+    "YouTubeThumbnail",
+    "YouTubeVideoThumbnails",
+    "YouTubeVideoSnippet",
+    "YouTubeVideo",
+    "YouTubeChannelThumbnails",
+    "YouTubeChannelRelatedPlaylists",
+    "YouTubeChannelContentDetails",
+    "YouTubeChannelSnippet",
+    "YouTubeChannel",
+]
+
 T = TypeVar("T")
 
 
-class YouTubeVideoThumbnail(BaseModel):
+class YouTubeThumbnail(BaseModel):
     """Model representing a video thumbnail."""
 
     url: str = Field(...)
@@ -20,11 +32,11 @@ class YouTubeVideoThumbnail(BaseModel):
 class YouTubeVideoThumbnails(BaseModel):
     """Model representing video thumbnails."""
 
-    default: YouTubeVideoThumbnail = Field(...)
-    medium: YouTubeVideoThumbnail = Field(...)
-    high: YouTubeVideoThumbnail = Field(...)
-    standard: YouTubeVideoThumbnail = Field(...)
-    maxres: YouTubeVideoThumbnail | None = Field(None)
+    default: YouTubeThumbnail = Field(...)
+    medium: YouTubeThumbnail = Field(...)
+    high: YouTubeThumbnail = Field(...)
+    standard: YouTubeThumbnail = Field(...)
+    maxres: YouTubeThumbnail | None = Field(None)
 
 
 class YouTubeVideoSnippet(BaseModel):
@@ -36,7 +48,7 @@ class YouTubeVideoSnippet(BaseModel):
     description: str = Field(...)
     thumbnails: YouTubeVideoThumbnails = Field(...)
     channel_title: str = Field(..., alias="channelTitle")
-    tags: list[str] = Field(...)
+    tags: list[str] = Field([])
     live_broadcast_content: LiveBroadcastContent = Field(
         ...,
         alias="liveBroadcastContent",
@@ -50,3 +62,48 @@ class YouTubeVideo(BaseModel):
 
     video_id: str = Field(..., alias="id")
     snippet: YouTubeVideoSnippet | None = None
+
+
+class YouTubeChannelThumbnails(BaseModel):
+    """Model representing channel thumbnails."""
+
+    default: YouTubeThumbnail = Field(...)
+    medium: YouTubeThumbnail = Field(...)
+    high: YouTubeThumbnail = Field(...)
+
+
+class YouTubeChannelRelatedPlaylists(BaseModel):
+    """Model representing related playlists of a channel."""
+
+    likes: str = Field(...)
+    uploads: str = Field(...)
+
+
+class YouTubeChannelContentDetails(BaseModel):
+    """Model representing content details of a channel."""
+
+    related_playlists: YouTubeChannelRelatedPlaylists = Field(
+        ...,
+        alias="relatedPlaylists",
+    )
+
+
+class YouTubeChannelSnippet(BaseModel):
+    """Model representing channel snippet."""
+
+    title: str = Field(...)
+    description: str = Field(...)
+    published_at: datetime = Field(..., alias="publishedAt")
+    thumbnails: YouTubeChannelThumbnails = Field(...)
+    default_language: str | None = Field(None, alias="defaultLanguage")
+
+
+class YouTubeChannel(BaseModel):
+    """Model representing a YouTube channel."""
+
+    channel_id: str = Field(..., alias="id")
+    snippet: YouTubeChannelSnippet | None = None
+    content_details: YouTubeChannelContentDetails | None = Field(
+        None,
+        alias="contentDetails",
+    )

src/async_python_youtube/models.py

@@ -0,0 +1,49 @@
+"""Oauth helpers for YouTube."""
+import aiohttp
+
+from async_python_youtube.helper import YOUTUBE_AUTH_TOKEN_URL, build_url
+from async_python_youtube.types import InvalidRefreshTokenError, UnauthorizedError
+
+__all__ = ["refresh_access_token"]
+
+
+async def refresh_access_token(
+    refresh_token: str,
+    app_id: str,
+    app_secret: str,
+    session: aiohttp.ClientSession | None = None,
+) -> tuple[str, str]:
+    """Refresh a user access token.
+
+    :param str refresh_token: the current refresh_token
+    :param str app_id: the id of your app
+    :param str app_secret: the secret key of your app
+    :param ~aiohttp.ClientSession session: optionally a active client session to be
+    used for the web request to avoid having to open a new one
+    :return: access_token, refresh_token
+    :raises ~async_python_youtube.types.InvalidRefreshTokenException: if refresh token
+    is invalid
+    :raises ~async_python_youtube.types.UnauthorizedException: if both refresh and
+    access token are invalid (e.g. if the user changes their password of the app gets
+    disconnected)
+    :rtype: (str, str)
+    """
+    param = {
+        "refresh_token": refresh_token,
+        "client_id": app_id,
+        "grant_type": "refresh_token",
+        "client_secret": app_secret,
+        "access_type": "offline",
+        "prompt": "consent",
+    }
+    url = build_url(YOUTUBE_AUTH_TOKEN_URL, {})
+    ses = session if session is not None else aiohttp.ClientSession()
+    async with ses.post(url, data=param) as result:
+        data = await result.json()
+    if session is None:
+        await ses.close()
+    if result.status == 400:
+        raise InvalidRefreshTokenError(data.get("error", ""))
+    if result.status == 401:
+        raise UnauthorizedError(data.get("error", ""))
+    return data["access_token"], refresh_token

src/async_python_youtube/oauth.py

@@ -0,0 +1,75 @@
+"""Type definitions."""
+from enum import Enum
+
+__all__ = [
+    "AuthScope",
+    "YouTubeAPIError",
+    "YouTubeAuthorizationError",
+    "InvalidRefreshTokenError",
+    "InvalidTokenError",
+    "UnauthorizedError",
+    "MissingScopeError",
+    "YouTubeBackendError",
+    "MissingAppSecretError",
+    "DeprecatedError",
+    "YouTubeResourceNotFoundError",
+    "ForbiddenError",
+]
+
+
+class AuthScope(Enum):
+    """Enum of authentication scopes."""
+
+    MANAGE = "https://www.googleapis.com/auth/youtube"
+    MANAGE_MEMBERSHIPS = (
+        "https://www.googleapis.com/auth/youtube.channel-memberships.creator"
+    )
+    FORCE_SSL = "https://www.googleapis.com/auth/youtube.force-ssl"
+    READ_ONLY = "https://www.googleapis.com/auth/youtube.readonly"
+    UPLOAD = "https://www.googleapis.com/auth/youtube.upload"
+    PARTNER = "https://www.googleapis.com/auth/youtubepartner"
+    PARTNER_AUDIT = "https://www.googleapis.com/auth/youtubepartner-channel-audit"
+
+
+class YouTubeAPIError(Exception):
+    """Base YouTube API exception."""
+
+
+class YouTubeAuthorizationError(YouTubeAPIError):
+    """Exception in the YouTube Authorization."""
+
+
+class InvalidRefreshTokenError(YouTubeAPIError):
+    """used User Refresh Token is invalid."""
+
+
+class InvalidTokenError(YouTubeAPIError):
+    """Used if an invalid token is set for the client."""
+
+
+class UnauthorizedError(YouTubeAuthorizationError):
+    """Not authorized to use this."""
+
+
+class MissingScopeError(YouTubeAuthorizationError):
+    """authorization is missing scope."""
+
+
+class YouTubeBackendError(YouTubeAPIError):
+    """When the YouTube API itself is down."""
+
+
+class MissingAppSecretError(YouTubeAPIError):
+    """When the app secret is not set but app authorization is attempted."""
+
+
+class DeprecatedError(YouTubeAPIError):
+    """If something has been marked as deprecated by the YouTube API."""
+
+
+class YouTubeResourceNotFoundError(YouTubeAPIError):
+    """If a requested resource was not found."""
+
+
+class ForbiddenError(YouTubeAPIError):
+    """If you are not allowed to do that."""