from datetime import date from typing import Any, Dict, Iterable, List, Optional import requests import config from constants import common from utils import handle_errors def _get_headers() -> Dict[str, str]: """Get headers for API request.""" return { "Authorization": config.DSPAPI.APP_KEY, "Content-Type": "application/json", "X-App-Slug": "apollo", "X-Client-Type": "job", "X-Client-Name": common.APP_NAME, } @handle_errors() def get(url: str, params: dict) -> Any: """Make get request. Args: url: URL. params: Request params. Returns: Parsed json result. """ response = requests.get(url, headers=_get_headers(), params=params, timeout=config.DSPAPI.TIMEOUT) return response.json() def get_playlists_tracks( playlist_id_list: Iterable[str], isrc_list: Optional[Iterable[str]] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, ) -> List[dict]: """Get Spotify playlists tracks. Args: playlist_id_list: Playlist ID list. isrc_list: ISRC list. start_date: Start date. end_date: End date. Returns: List of tracks. """ url = f"{config.DSPAPI.HOST}/api/delphi/public/track-positions/playlists" offset = 0 result = [] while True: data = get( url, params={ "offset": offset, "limit": config.DSPAPI.PAGE_SIZE, "playlist_id": ",".join(f"spotify_{i}" for i in playlist_id_list), **( { "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "sort_by": "date", "sort_order": "asc", } if start_date and end_date else {} ), **({"isrc": ",".join(isrc_list)} if isrc_list else {}), } ) result += data.get("items", []) if data.get("next"): offset += config.DSPAPI.PAGE_SIZE else: break return result