""" YouTube base API Client. Uses aiogoogle to make requests to YouTube APIs, instead of the official Google API client for Python. This is because as of Jan 2024, the official client doesn't support asynchronous requests and thus isn't optimal for async applications. """ import asyncio import itertools import json from abc import ABC, abstractmethod from typing import Any, Callable, Type from aiohttp.client_exceptions import ClientOSError, ServerDisconnectedError from ... import logger from ...constants import YtCid from . import decorators, exceptions, helpers from .constants import RequestPayload from .typings import ApiScopes logger = logger.new_logger(__name__) class BaseClient(ABC): """Orchard proprietary base Client for the YouTube Content ID API. Properties: request_count: The number of requests made to the API since the client was initialized. """ # Common aiogoogle client instance for all subclasses. _aiogoogle = None # OVERRIDE THE FOLLOWING PROPERTIES IN SUBCLASSES IF NEEDED: # Max. number of concurrent requests to the API to prevent rate limiting. # Override in subclasses if needed. _api_max_concurrent_requests: int = 20 # How many items to fetch per request when using a bulk endpoint # As of Dec 2023, API has a max. limit of 50 items per bulk request. _max_bulk_items_per_chunk: int = 50 # The maximum number of items in a paginated request. # As of Dec 2023, API has a max. limit of 50 items per request. _max_items_per_page: int = 50 @classmethod @abstractmethod def _init_lock(cls) -> asyncio.Lock: """Async lock to control the initialization of the client.""" @classmethod @abstractmethod def _semaphore(cls) -> asyncio.Semaphore: """Async semaphore to limit the number of concurrent requests to the API. Use _api_max_concurrent_requests to set the number of concurrent requests for the specific API. """ @classmethod @abstractmethod def _api(cls) -> asyncio.Semaphore: """Async semaphore to limit the number of concurrent requests to the API.""" @classmethod @abstractmethod def _api_name(cls) -> str: """API name as defined in the aiogoogle library, e.g. 'youtubePartner'.""" @classmethod @abstractmethod def _api_version(cls) -> str: """API version as defined in the aiogoogle library, e.g. 'v1'.""" @classmethod @abstractmethod def _scopes(cls) -> ApiScopes: """API scopes as defined in the aiogoogle library, e.g. ("https://www.googleapis.com/auth/youtubepartner",). """ def __init__(self): """Initialize the client.""" self._request_count: int = 0 self._request_count_successful: int = 0 @property def request_count(self) -> int: """Return the number of requests made to the API since the client was initialized. Includes both successful and failed requests. """ return self._request_count @property def request_count_successful(self) -> int: """Return the number of successful requests made to the API since the client was initialized. Successful requests are those that returned a response without raising an exception. """ return self._request_count_successful @property def credentials(self) -> dict[str, Any] | None: """Return the current credentials used by the client, with the private key truncated for security reasons. Those are merely for logging purposes. If the client hasn't been initialized yet, this will return None. """ try: current_credentials = self._aiogoogle.service_account_creds except AttributeError: return None obfuscated_credentials = { k: (f"{v[:40]}***" if k == "private_key" else v) for k, v in current_credentials.items() if not k.startswith("'") } return obfuscated_credentials @decorators.requires_init async def init(self): """Manual initialization of the client. Can be useful for async initialization to have the client ready before the first request. """ 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 the 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. Will 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. """ kwargs[RequestPayload.MAX_RESULTS] = limit or self._max_items_per_page 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 available 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