""" Youtube Content ID API Client. """ import asyncio import itertools import json from asyncio import Lock, Semaphore from typing import Any, Callable, Iterable, Type from aiohttp.client_exceptions import ClientOSError, ServerDisconnectedError from .... import logger from ....constants import YtCid from ....typings import ISRC, AssetID, VideoID from .. import constants, exceptions, helpers from ..client import BaseClient from ..decorators import requires_init from ..typings import ApiScopes from . import models 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.CONTENT_ID _api_version: str = "v1" _scopes: ApiScopes = (constants.Scope.CONTENT_ID,) _api = None _init_lock = Lock() _semaphore = Semaphore(BaseClient._api_max_concurrent_requests) @requires_init async def asset_relationships_list( self, asset_id: AssetID, **kwargs, ) -> models.AssetRelationships: """Retrieve asset relationships for a given asset. Args: asset_id: The ID of the asset to retrieve relationships for. KwArgs: kwargs: Additional arguments to pass to the API. Any keyword arguments admitted by the API will be accepted here. Returns: An AssetRelationships object containing the asset's parent and/or child assets, if any. """ asset_relationships = models.AssetRelationships(id=asset_id) assets = await self._dispatch_paginated( self._api.assetRelationships.list, assetId=asset_id, **kwargs ) if not assets: return asset_relationships children = asset_relationships.children parents = asset_relationships.parents for asset in assets: parent_asset_id = asset.get(YtCid.PARENT_ASSET_ID) if asset_id == parent_asset_id: children.append(asset[YtCid.CHILD_ASSET_ID]) else: parents.append(parent_asset_id) return asset_relationships @requires_init async def asset_search_list(self, **kwargs) -> list[models.AssetLookupResult]: """Lookup assets. Use the 'asset_search_list_isrc' method instead if you want to lookup assets by ISRC. https://developers.google.com/youtube/partner/reference/rest/v1/assetSearch/list KwArgs: kwargs: Arguments to pass to the API. Any keyword arguments admitted by the API will be accepted here. """ # ISRCS parameter can't be used here because it requires a bulk # operation. if YtCid.ISRCS in kwargs: raise ValueError( "The 'isrcs' parameter is not supported." "Use 'asset_search_list_isrc' method instead." ) asset_lookup_results = await self._get_list_objects( models.AssetLookupResult, self._api.assetSearch.list, **kwargs ) return asset_lookup_results @requires_init async def asset_search_list_isrc( self, isrcs: list[ISRC] | tuple[ISRC, ...] | set[ISRC], **kwargs ) -> list[models.AssetLookupResult]: """Lookup assets by ISRC. https://developers.google.com/youtube/partner/reference/rest/v1/assetSearch/list Args: isrcs: The ISRCs of the assets to retrieve. KwArgs: kwargs: Additional arguments to pass to the API. Any keyword arguments admitted by the API will be accepted here. """ asset_lookup_results = await self._batch_request( self._api.assetSearch.list, models.AssetLookupResult, YtCid.ISRCS, {YtCid.ISRCS: isrcs, **kwargs}, ) return asset_lookup_results @requires_init async def assets_list( self, asset_ids: list[AssetID] | tuple[AssetID, ...] | set[AssetID], fetch_ownership: str = YtCid.EFFECTIVE, fetch_metadata: str = YtCid.EFFECTIVE, fetch_match_policy: str = YtCid.EFFECTIVE, **kwargs, ) -> list[models.Asset]: """Retrieve metadata for a list of assets in a single request. https://developers.google.com/youtube/partner/reference/rest/v1/assets/list Args: asset_ids: The IDs of the assets to retrieve. fetch_ownership: Whether to fetch ownership data. Must be one of "mine", "effective", or None. fetch_metadata: Whether to fetch metadata. Must be one of "mine", "effective", or None. fetch_match_policy: Whether to fetch match policy. Must be one of "mine", "effective", or None. KwArgs: kwargs: Additional arguments to pass to the API. Any keyword arguments admitted by the API will be accepted here. """ id_key = YtCid.ID params = { **kwargs, id_key: asset_ids, YtCid.FETCH_OWNERSHIP: fetch_ownership, YtCid.FETCH_METADATA: fetch_metadata, YtCid.FETCH_MATCH_POLICY: fetch_match_policy, } assets = await self._batch_request( self._api.assets.list, models.Asset, id_key, params ) return assets @requires_init async def metadata_history(self, asset_id: AssetID) -> list[models.MetadataHistory]: """Retrieve metadata history for an asset. Args: asset_id: The ID of the asset to retrieve metadata history for. """ history = await self._get_list_objects( models.MetadataHistory, self._api.metadataHistory.list, assetId=asset_id ) return history @requires_init async def references_list( self, asset_id: AssetID | Iterable[AssetID] ) -> list[models.Reference]: """Retrieve references for one or more assets. Args: asset_id: The ID of the asset(s) to retrieve references for. Returns: A single list of references for all the assets. The references can be related to specific assets by their properties. """ if isinstance(asset_id, str): asset_id = [asset_id] model = models.Reference api_func = self._api.references.list tasks = ( self._get_list_objects(model, api_func, assetId=asset_id) for asset_id in set(asset_id) ) flattened_references = [ ref for batch in await asyncio.gather(*tasks) for ref in batch ] return flattened_references @requires_init async def claims_by_asset_id( self, asset_id: AssetID, limit: int = None ) -> list[models.Claim]: """Retrieve all claims for a given Asset ID. Args: asset_id: The ID of the asset to retrieve claims for. limit: The maximum number of claims to retrieve. If None, all available claims will be retrieved. """ return await self._claims_list(asset_id=asset_id, limit=limit) @requires_init async def claims_by_video_id( self, video_id: VideoID, limit: int = None ) -> list[models.Claim]: """Retrieve all claims for a given Video ID. Args: video_id: The ID of the video to retrieve claims for. limit: The maximum number of claims to retrieve. If None, all available claims will be retrieved. """ return await self._claims_list(video_id=video_id, limit=limit) async def _claims_list( self, asset_id: AssetID = None, video_id: VideoID = None, limit: int = None ) -> list[models.Claim]: """Retrieve all claims for a given asset or video ID. Args: asset_id: The ID of the asset to retrieve claims for. video_id: The ID of the video to retrieve claims for. limit: The maximum number of claims to retrieve. If None, all available claims will be retrieved. Raises: ValueError: If neither asset_id nor video_id is provided, or if both are provided. """ if sum(1 for _ in [asset_id, video_id] if _ is not None) != 1: raise ValueError( "Exactly one of 'asset_id' or 'video_id' must be provided." ) model = models.Claim api_func = self._api.claims.list claims = await self._get_list_objects( model, api_func, assetId=asset_id, videoId=video_id, limit=limit ) return claims async def _batch_request( self, api_endpoint_func: Callable, resp_obj_type: Type, id_key: str, api_params: dict[str, Any], ): """Make a batch request to the API. This is for endpoints accepting a list of IDs or similar as input, up to the maximum allowed by the API (50 as of Dec 2023). If more than 50 items are passed, the request will be split into multiple requests, which will be executed concurrently. The results will be merged into a single list. Args: api_endpoint_func: A callable that returns an API endpoint response. resp_obj_type: The type of the response objects which will be constructed from the API response and included in the returned list. id_key: The name of the ID parameter in the API, the values of which will be split into batches. api_params: Parameters to pass to the API. """ asset_id_batches = ( ",".join(batch) for batch in itertools.batched( api_params[id_key], self._max_bulk_items_per_chunk ) ) tasks = [] for asset_id_batch in asset_id_batches: def create_task( _asset_id_batch=asset_id_batch, ): # capture current value of asset_id_batch for this iteration! # Not capturing it can cause issues where each task shares the same # reference to the same asset_id_batch variable; this can cause # duplicate and missing values in the final result. api_endpoint_func_params = {**api_params, id_key: _asset_id_batch} return asyncio.create_task( self._get_list_objects( resp_obj_type, api_endpoint_func, **api_endpoint_func_params ) ) tasks.append(create_task()) worker = helpers.batch_worker() task_results = await asyncio.gather(*map(worker, tasks)) results = list(itertools.chain.from_iterable(task_results)) return results async def _get_list_objects( self, obj_factory: Callable, func: Callable, *args, limit: int = None, **kwargs ) -> list: """Get a list of objects from a list API endpoint response. Args: func: A callable that returns a list API endpoint response. obj_factory: A callable that accepts keyword arguments and returns an object. Wll be used to construct the objects from the response. args: Positional arguments to pass to the API. limit: The maximum number of items to collect. If None, all available items will be collected. kwargs: Keyword arguments to pass to the API. Returns: A list of objects constructed by the given factory. """ items = await self._dispatch_paginated(func, *args, limit=limit, **kwargs) objects = [helpers.to_response_object(item, obj_factory) for item in items] return objects async def _dispatch(self, func: Callable, retries: int = 10) -> Any: """Dispatch an async request to the API. Args: func: A callable that returns an API response. retries: The number of retries to attempt if the request fails. Only applies to certain exceptions. Returns: The API response. """ try: async with self._semaphore: async with self._aiogoogle: request = func() self._request_count += 1 logger.debug( f"Dispatching request to YouTube Content ID API. URL: {request.url}" ) responses = await self._aiogoogle.as_service_account(request) self._request_count_successful += 1 logger.debug("Response received from YouTube Content ID API.") except Exception as ex: # Handle specific exceptions based on the response status code. # Retries the request if it's a recoverable error. responses = await self._handle_exception(ex, func=func, retries=retries) return responses async def _dispatch_paginated( self, func: Callable, *args, limit: int = None, **kwargs ) -> list[dict]: """Make a paginated request to the API. Args: func: A callable that returns an API response. args: Positional arguments to pass to the API. limit: The maximum number of items to collect. If None, all avaliable items will be collected. kwargs: Keyword arguments to pass to the API. Returns: A list of items collected from the paginated response. No other data of the individual responses is returned. """ page_token = kwargs.pop(YtCid.PAGE_TOKEN, None) collected_items = [] while len(collected_items) < limit if limit else True: response = await self._dispatch( lambda: func(*args, pageToken=page_token, **kwargs) ) items = response.get(YtCid.ITEMS, []) if not items: return collected_items collected_items.extend(items) page_token = response.get(YtCid.NEXT_PAGE_TOKEN) if not page_token: break return collected_items[:limit] async def _handle_exception( self, ex: Exception, *, func: Callable, retries: int, retry_after_secs: int = 10 ) -> Any: """Handle exceptions raised during API requests. Args: ex: The exception raised during the request, as returned by the aiogoogle library. func: The function that was being executed when the exception was raised. retries: The number of retries left for the request. If the request fails due to a recoverable error, it will be retried up to this number of times in a recursive manner. """ if hasattr(ex, "res"): status_code = ex.res.status_code if status_code == 404: raise exceptions.NotFound() from ex if status_code == 400: logger.error( "There was a problem with the request. The " "following credentials were used (truncated " "private_key for security purposes):\n" "{}", json.dumps(self.credentials, indent=2), ) raise exceptions.BadRequest() from ex if status_code in {429, 500}: if retries > 0: retries -= 1 if status_code == 500: logger.warning( "Received HTTP 500 response. Retrying in {}s...", retry_after_secs, ) await asyncio.sleep(retry_after_secs) return await self._dispatch(func, retries=retries) if status_code == 429: # This should only happen if the API daily rate limits are exceeded, # so further retries are pointless. raise exceptions.APIQuotaExceeded() from ex raise exceptions.InternalServerError() from ex raise ex aiohttp_exceptions = (ServerDisconnectedError, ClientOSError) if isinstance(ex, aiohttp_exceptions): if retries > 0: retries -= 1 if isinstance(ex, ServerDisconnectedError): logger.warning("Server disconnected. Retrying...") if isinstance(ex, ClientOSError): logger.warning("Client OS error. Retrying...") return await self._dispatch(func, retries=retries) raise ex raise ex