import time from typing import List import allure from attr import asdict from ui_automation_framework.utils import CoreConfig from app.src.api.models.portal_search_response import PortalPlaylistsSearch, PortalSearchResponse from app.src.api.models.user_settings import UserSettings, UserSettingsGetResponse class ApiClient: def __init__(self, session, host_endpoint=CoreConfig.TARGET_API_URL, timeout=60): self.host_endpoint = host_endpoint self.timeout = int(timeout) self.session = session self.session.headers["Cache-Mode"] = "ignore" @allure.step("Getting CA track container data") def get_ca_track_container_data(self, vendor, isrc, start_date, end_date, market): api_string = "/dsp-api/consumer_analytics/v1/playlists-summary" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, isrc=isrc, start_date=start_date, end_date=end_date, market=market), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting total number of records for the apple track playlists in portal endpoint") def get_apple_playlists_positions_portal(self, market, isrc, limit, offset): resp = self.session.get( self.host_endpoint + "/applemusic/playlistswithtrack", timeout=self.timeout, params=dict(isrc=isrc, limit=limit, country=market, offset=offset), ) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting charts summary for spotify for portal endpoint spotifycharts/tracksummary/{isrc}") def get_charts_summary_spotify_portal(self, isrc): api_string = "/spotifycharts/tracksummary/{}".format(isrc) resp = self.session.get(self.host_endpoint + api_string, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting charts summary for aplle for portal endpoint applemusic/charts/tracksummary") def get_charts_summary_apple_portal(self, isrc): api_string = "/applemusic/charts/tracksummary" resp = self.session.get(self.host_endpoint + api_string, params=dict(isrc=isrc), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting latest date from /spotify-consumer-analytics-compat/latest-date") def get_latest_date_portal(self, isrc): api_string = "/spotify-consumer-analytics-compat/latest-date?" resp = self.session.get(self.host_endpoint + api_string, params=dict(isrc=isrc), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting is sony flag from /sonyalbums/issony?") def get_is_sony_portal(self, *parameters): api_string = "/sonyalbums/issony?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting chart placement for portal spotify track from /spotifycharts/tracksummary/") def get_chart_placement_portal_spotify(self, isrc): api_string = "/spotifycharts/tracksummary/{}".format(isrc) resp = self.session.get(self.host_endpoint + api_string, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting chart placement for portal apple track from /applemusic/charts/tracksummary?") def get_chart_placement_portal_apple(self, *parameters): api_string = "/applemusic/charts/tracksummary?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting Portal Apple playlists data from /applemusic/playlistswithtrack?") def get_apple_playlists_data_portal(self, isrc, country=None, limit=None, offset=None): # api_string = "/applemusic/playlistswithtrack" api_string = "/gate-api/v1/playlists/apple/by-track/" resp = self.session.get(self.host_endpoint + api_string, params=dict(isrc=isrc, country=country, limit=limit, offset=offset), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response def get_apple_demographics_consumer_analytics(self, isrc, start_date, end_date, market, token): headers = {"authorization": "Bearer {}".format(token)} api_string = "/apple-consumer-analytics/track-demographics?startDate={}&endDate={}&isrc={}".format(start_date, end_date, isrc) resp = self.session.get(self.host_endpoint + api_string, headers=headers) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting chart data from /gate-api/v1/playlists/spotify/tracks/") def get_sonytracks_analyze(self, playlisturi, region): api_string = "/gate-api/v1/playlists/spotify/tracks/" resp = self.session.get(self.host_endpoint + api_string, params=dict(playlistUri=playlisturi, region=region), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting chart data from /gate-api/v1/playlists/apple/tracks/") def get_applemusic_playlists(self, playlist_id, market): api_string = "/gate-api/v1/playlists/apple/tracks/?playlist_id={}".format(playlist_id) resp = self.session.get(self.host_endpoint + api_string, params=dict(country=market), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting search history from /spotifysearchhistory/auth0") def get_spotify_search_history_auth0(self, client_id, *parameters): api_string = "/spotifysearchhistory/auth0%{}?limit=21".format(client_id) params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} return response @allure.step("Getting user's settings from /user-data-api/v1/settings/") def get_user_data_api_v1_settings(self, headers=None): api_string = "/user-data-api/v1/settings/" if not headers: headers = {"X-App-Slug": "apollo"} else: headers = headers return self.session.get(self.host_endpoint + api_string, headers=headers, timeout=self.timeout).json() return UserSettingsGetResponse(**self.session.get(self.host_endpoint + api_string, headers=headers, timeout=self.timeout).json()) def put_user_data_api_v1_settings(self, user_settings: UserSettings): headers = {"X-App-Slug": "apollo"} api_string = "/user-data-api/v1/settings/" return UserSettingsGetResponse( **self.session.put( self.host_endpoint + api_string, headers=headers, json=dict(timestamp=time.time() * 1000, data=asdict(user_settings)), ).json() ) def get_search_results_portal(self, query, item_type: List, limit=20): """ :param query: :param item_type: could be track,artist, album """ resp = self.session.get( self.host_endpoint + "/vendor-api/spotify/v1/search", params=dict(query=query, item_type=",".join(item_type), limit=limit), ) return PortalSearchResponse(**resp.json()) def get_playlists_apple(self, query, limit=20): return PortalPlaylistsSearch(**self.session.get(self.host_endpoint + "/applemusic/search", params=dict(search=query, limit=limit)).json()) def get_playlists_spotify(self, query, limit=20): return PortalPlaylistsSearch(**self.session.get(self.host_endpoint + "/spotifysearch", params=dict(search=query, limit=limit)).json())