""" Youtube Data API Client. """ from asyncio import Lock, Semaphore from .... import logger from ....typings import ChannelID from .. import constants from ..client import BaseClient from ..decorators import requires_init from ..typings import ApiScopes from . import models from .constants import ResourceType logger = logger.new_logger(__name__) class Client(BaseClient): """Orchard proprietary Client for the YouTube Content ID API. Properties: request_count: The number of requests made to the API since the client was initialized. """ _api_name: str = constants.YouTubeAPI.DATA _api_version: str = "v3" _scopes: ApiScopes = (constants.Scope.DATA,) _api = None _init_lock = Lock() _semaphore = Semaphore(BaseClient._api_max_concurrent_requests) @requires_init async def _search_list(self, **kwargs) -> list[models.SearchResult]: """Generic search list method for the YouTube Data API. https://developers.google.com/youtube/v3/docs/search """ lookup_results = await self._get_list_objects( models.SearchResult, self._api.search.list, **{ "part": "snippet", # Mandatory, always this constant **kwargs, }, ) return lookup_results @requires_init async def channel_videos_list( self, channel_id: ChannelID, **kwargs ) -> list[models.Video]: """Get the metadata of all videos in a channel. Args: channel_id: The ID of the channel to search for, e.g. "UC_x5XG1OV2P6uZZ5FSM9Ttw". """ search_results = await self._search_list( channelId=channel_id, type=ResourceType.VIDEOS, **kwargs, ) videos = [ models.Video(**{**r.snippet.__dict__, "video_id": r.id.video_id}) for r in search_results ] return videos