import asyncio import itertools from datetime import date, datetime from typing import Any, Dict, List, Optional from server import config from server.client.base.delphi_base_client import DelphiBaseClient from server.constants import dsp from server.constants.core import ALL from server.constants.delphi.tiktok import TiktokTrackAnalyticsBreakdown from server.constants.dsp import DSP_URL_MAPPING from server.schemas.delphi.charts.analytics import DelphiChartsAnalytics from server.schemas.delphi.charts.v3 import DelphiChartsV3 from server.schemas.delphi.deserializers.regions import DelphiRegionsParams from server.schemas.delphi.deserializers.streams.dates import DelphiStreamDatesParams from server.schemas.delphi.deserializers.streams.v3 import DelphiStreamsV3 from server.schemas.delphi.deserializers.videos.analytics import DelphiVideoAnalyticsParams from server.schemas.delphi.deserializers.videos.misc import DelphiVideosParams from server.schemas.delphi.playlists import DelphiPlaylistsFollowers from server.schemas.delphi.tracks.charts import DelphiTracksCharts from server.schemas.delphi.tracks.tiktok.analytics import DelphiTiktokTracksAnalyticsV3RequestSchema from server.schemas.delphi.videos.charts.positions import ( DelphiVideoPositionsChartsSummaryV3, DelphiVideoPositionsChartsV3, ) from server.utils.args.prepare import prepare_list_arg from server.utils.cleaners import clean, clean_video_analytics from server.utils.client.delphi.request_per_item import request_per_item, request_per_item_many from server.utils.combine import add_by_keys from server.utils.delphi.counters import count_totals from server.utils.delphi.misc import ( request_related_isrc, resolve_streams_include, update_breakdowns_for_totals, fill_skipped_dates, ) from server.utils.delphi.summarize.chart_analytics import sum_chart_analytics from server.utils.delphi.summarize.tiktok import sum_tiktok_tracks_analytics from server.utils.pagination.delphi_pagination import delphi_pagination as pagination class DelphiClient(DelphiBaseClient): """Delphi API client.""" @request_per_item("dsp_list", default_value=dsp.ALL_VENDORS, enable_arg_name="request_per_dsp") @pagination() @clean() async def get_streams( self, start_date: date or str, end_date: date or str, isrc_list: List[str] or None = None, playlist_id_list: List[str] or None = None, artist_id: str or None = None, track_id_list: List[str] or None = None, dsp_list: List[str] or None = None, country_code_list: List[str] or None = None, group_by: List[str] or None = None, subset: str or None = None, agg_by: str or None = None, include: List[str] or None = None, limit_range: str or None = None, offset: int or None = None, limit: int or None = None, cursor: str or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get streams with additional data. Args: start_date: start date of the range. end_date: end date of the range. isrc_list: list of isrc to get streams for, None for all ISRC. playlist_id_list: list of playlist ID, None for all playlists. artist_id: artist ID, None for all. track_id_list: list of track ID, None for all. dsp_list: list of vendors to get data for. country_code_list: list of markets to get data for, pass None for all markets. group_by: list of keys to group response by, check StreamsGroupBy for available options. subset: aggregate by this field, now only 'playlists' option is available. agg_by: aggregate by this field, now only 'isrc' and 'artist' options are available. include: extra information to be included in response, check StreamsInclude for available options. limit_range: pagination limit to specify the number of days to include in a single response page. offset: pagination offset. limit: pagination limit. cursor: pagination parameter, value from the previous response next_cursor field. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. Returns: dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ # rewrite inner include types like 'sources', 'engagement' include = resolve_streams_include(include) params = DelphiStreamsV3.Base().dump(locals()) return await self.get("streams", params) @staticmethod def format_streams(data: List[dict], remove_dsp: bool = True, dsp_prefix: str = None) -> List[dict]: """Format Delphi streaming data. Args: data: Streams data. remove_dsp: Remove dsp field. dsp_prefix: Optional prefix for streams_info data, skips if passed. Returns: Cleaned streams. """ for item in data: item_dsp = item["dsp"] prefix = dsp_prefix or item_dsp streams_info_key = f"{prefix}_streams_info" item.update(item[streams_info_key]) del item[streams_info_key] if remove_dsp: del item["dsp"] elif dsp_prefix and item_dsp.startswith(dsp_prefix): item["dsp"] = item_dsp[len(dsp_prefix) :] return data @pagination() async def get_first_stream_dates( self, isrc_list: List[str], dsp_list: List[str] or None = None, offset: int or None = None, limit: int or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> List[dict]: """Get first stream dates. Args: isrc_list: list of isrc to get streams for. dsp_list: list of vendors to get data for, None for all. offset: pagination offset. if key in item and item[key] is None: del item[key] limit: pagination limit. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. Returns: First track stream dates. """ params = DelphiStreamDatesParams().dump(locals()) return await self.get("stream-dates", params) async def get_latest_stream_dates(self) -> List[dict]: """Get latest completed stream dates. Returns: Latest completed dates. """ result = await self.get("data-health/completeness") return result["items"] async def get_latest_stream_date_vendor(self, vendor: str = dsp.SPOTIFY) -> date: """Get latest date of streams for specific vendor. Args: vendor: Vendor name. """ result = await self.get_latest_stream_dates() latest_date = next(i["updated_date"] for i in result if i["dsp"] == vendor) return datetime.strptime(latest_date, "%Y-%m-%d").date() async def get_min_latest_date(self, dsp_list: List[str] = None) -> date: """Get min latest available metrics date for vendors. Args: dsp_list: Vendor names. """ result = await self.get_latest_stream_dates() dsp_list = dsp_list or [] return min(datetime.strptime(i["updated_date"], "%Y-%m-%d").date() for i in result if i["dsp"] in dsp_list) async def get_latest_dates(self, dsp_list: List[str] = None) -> Dict[str, date]: """Get latest available metrics dates per vendor. Args: dsp_list: Vendor names. """ result = await self.get_latest_stream_dates() dsp_list = dsp_list or [] return { i["dsp"]: datetime.strptime(i["updated_date"], "%Y-%m-%d").date() for i in result if i["dsp"] in dsp_list } async def get_streams_extended( self, isrc_items_list: List[str] or str, dsp: str, country_code_list: List[str] or None = None, include: List[str] or None = None, combine_isrc: bool = False, ) -> List[Dict[str, Any]]: """Get demographic/saves/skips data. Args: isrc_items_list: A list of dictionaries with fields "isrc", "start_date", "end_date". dsp: Vendor to get data for. country_code_list: List of markets to get data for. Pass None to get all available markets. include: Include additional fields like demographics or skips. combine_isrc: Flag, if True method returns summed up data for all passed isrc. Returns: A list of items for different isrc and markets. """ # Delphi team recommended to request data by one isrc in parallel instead list for better performance. tasks = [ self.get_streams( start_date=isrc_item["start_date"], end_date=isrc_item["end_date"], isrc_list=[isrc_item["isrc"]], dsp_list=[dsp], country_code_list=country_code_list, include=include, ) for isrc_item in isrc_items_list ] responses = await asyncio.gather(*tasks) if combine_isrc: return self.combine_isrc(responses, ",".join(i["isrc"] for i in isrc_items_list)) return list(itertools.chain(*responses)) @staticmethod def combine_isrc(results: List[List[Dict[str, Any]]], isrc_str: str) -> List[Dict[str, Any]]: """Sum up data for all isrcs in the result. Args: results: list of result for each isrc. isrc_str: string of comma-separated isrc. Returns: list of items for different markets with summed up data for all isrc. """ combined_data = dict() excluded_keys = ("artist_id", "country_code", "date", "dsp", "isrc", "playlist_id", "track_id") for isrc_result in results: for market_item in isrc_result: market = market_item["country_code"] combined = combined_data.get(market) if combined: add_by_keys(combined, market_item, excluded_keys=excluded_keys, none_as=0) else: market_item["isrc"] = isrc_str combined_data[market] = market_item return list(combined_data.values()) @request_per_item_many(("isrc", "video_id")) @pagination(page_size=config.DELPHI_VIDEO_ANALYTICS_PAGE_SIZE, in_parallel=True) @clean_video_analytics() async def get_video_analytics( self, start_date: date or str, end_date: date or str, isrc: List[str] or None = None, dsp: str or None = None, country_code_list: List[str] or None = None, video_id: List[str] or None = None, expand_to: List[str] or None = None, content_type: List[str] or None = None, group_by: List[str] or None = None, only: List[str] or None = None, limit_range: str or None = None, offset: int or None = None, limit: int or None = None, cursor: str or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get youtube video streams with additional data. Args: start_date: start date of the range. end_date: end date of the range. isrc: ISRC to get streams for, None for all ISRC, the bulk_request decorator allows to pass a list of ISRC. dsp: DSP / vendor, now supports only youtube. country_code_list: list of markets to get data for, pass None for all markets. video_id: Video ID or None for all. expand_to: It supports only one value ("related_isrcs") right now, which is used to return all product ISRCs with streaming data by setting only one of them. content_type: Content type filter. group_by: list of keys to group response by, check StreamsGroupBy for available options. only: select fields to get in response, None - all. limit_range: pagination limit to specify the number of days to include in a single response page. offset: pagination offset. limit: pagination limit. cursor: pagination parameter, value from the previous response next_cursor field. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. Returns: dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ params = DelphiVideoAnalyticsParams().dump(locals()) return await self.get("video-analytics", params) @pagination(page_size=config.DELPHI_VIDEOS_PAGE_SIZE) async def get_videos( self, artist_id: str or None = None, channel_id: str or None = None, isrc_list: List[str] or None = None, track_id_list: List[str] or None = None, dsp: str or None = None, expand_to: List[str] or None = None, content_type: List[str] or None = None, offset: int or None = None, limit: int or None = None, sort_by: str or None = None, sort_order: str or None = None, views: int or None = None, ) -> Dict[str, Any]: """Get youtube videos metadata. Args: artist_id: Artist ID. channel_id: Youtube channel ID. isrc_list: A list of isrc to get streams for, None for all ISRC. track_id_list: A list of track ID. dsp: DSP / vendor, now supports only youtube. expand_to: It supports only one value ("related_isrcs") right now, which is used to return all product ISRCs with streaming data by setting only one of them. content_type: Content type filter. offset: pagination offset. limit: pagination limit. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. views: Views count filter. Returns: dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ params = DelphiVideosParams().dump(locals()) response = await self.get("videos", params) if views: response["items"] = [i for i in response.get("items", {}) if (i.get("views") or 0) >= views] return response async def get_video(self, video_id: str) -> Dict[str, Any]: """Get single youtube video metadata. Args: video_id: Youtube video ID. Returns: Video ID metadata. """ return await self.get(f"videos/{video_id}") @pagination(page_size=config.DELPHI_CHARTS_PAGE_SIZE) async def get_charts( self, chart_group: str or None = None, country_code: str or None = None, dsp_list: List[str] or None = None, limit: int or None = None, offset: int or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get charts data. Args: chart_group: Identifier for a chart group (category). country_code: Typically a lower-case two letter code for the country. dsp_list: list of vendors to get data for. limit: The maximum number of items to return in a single request (i.e.: a single page). offset: The number of items to offset (aka skip) for pagination. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. Returns: dictionary with "count" and "items". """ params = DelphiChartsV3.RequestSchema().dump(locals()) return await self.get("charts", params) async def get_chart_positions_dates( self, dsp_list: List[str] = None, chart_group_list: List[str] = None ) -> List[dict]: """Get available dates ranges per DSP / group. Args: dsp_list: DSP list filter. chart_group_list: Chart group filter. Returns: Processed dates ranges. """ result = await self.get("data-health/chart-positions") result = result["items"] if dsp_list: result = [i for i in result if i.get("dsp") in dsp_list] if chart_group_list: result = [i for i in result if i.get("chart_group") in chart_group_list] return result @request_per_item("video_id_list") @pagination(page_size=config.DELPHI_CHARTS_PAGE_SIZE) async def get_video_positions_charts_summary( self, artist_id: str or None = None, chart_group: str or None = None, chart_id: str or None = None, content_type_list: List[str] or None = None, country_code: str or None = None, dsp: str or None = None, expand_to_list: List[str] or None = None, include_list: List[str] or None = None, is_sony: bool or None = None, isrc_list: List[str] or None = None, video_id_list: List[str] or None = None, offset: int or None = None, limit: int or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get summary video data. Args: artist_id: Artist ID. chart_group: Identifier for a video chart group (category). chart_id: Chart ID (group + country code?). content_type_list: Content Type list, e.g.: partner_uploaded, premium_ugc, etc. country_code: Typically a lower-case two letter code for the country. dsp: The slug name of a video DSP. expand_to_list: Providing expand_to=related_isrcs will expand the query to include data from related ISRCs by using the parent Product Family. If this is provided, an isrc is required to be sent with the request. include_list: Include additional data like the Video object in each response item. is_sony: Filter for entities that have/have not been claimed by SME. isrc_list: Array of International Standard Recording Code (ISRC) numbers. video_id_list: Video ID list filter. offset: The number of items to offset (aka skip) for pagination. limit: The maximum number of items to return in a single request (i.e.: a single page). sort_by: Field name existing within objects in items by which to sort the results (dot notation). sort_order: Direction to sort the data. Default: desc if sort_by provided. Returns: Dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ params = DelphiVideoPositionsChartsSummaryV3.RequestSchema().dump(locals()) return await self.get("video-positions/charts/summary", params) @request_per_item("video_id_list") @pagination(page_size=config.DELPHI_CHARTS_PAGE_SIZE) async def get_video_positions_charts( self, start_date: date, end_date: date, artist_id: str or None = None, chart_group: str or None = None, chart_id: str or None = None, content_type_list: List[str] or None = None, country_code: str or None = None, dsp: str or None = None, expand_to_list: List[str] or None = None, include_list: List[str] or None = None, is_sony: bool or None = None, isrc_list: List[str] or None = None, video_id_list: List[str] or None = None, limit_range: str or None = None, cursor: str or None = None, offset: int or None = None, limit: int or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get summary video data. Args: start_date: The earliest date to include in the query range. end_date: The latest date to include in the query. artist_id: Artist ID. chart_group: Identifier for a video chart group (category). chart_id: Chart ID (group + country code?). content_type_list: Content Type list, e.g.: partner_uploaded, premium_ugc, etc. country_code: Typically a lower-case two letter code for the country. dsp: The slug name of a video DSP. expand_to_list: Providing expand_to=related_isrcs will expand the query to include data from related ISRCs by using the parent Product Family. If this is provided, an isrc is required to be sent with the request. include_list: Include additional data like the Video object in each response item. is_sony: Filter for entities that have/have not been claimed by SME. isrc_list: Array of International Standard Recording Code (ISRC) numbers. video_id_list: Video ID list filter. limit_range: Pagination limit to specify the number of days to include in a single response page in the format: days:28. cursor: This pagination parameter should be the value from a previous response's next_cursor. offset: The number of items to offset (aka skip) for pagination. limit: The maximum number of items to return in a single request (i.e.: a single page). sort_by: Field name existing within objects in items by which to sort the results (dot notation). sort_order: Direction to sort the data. Default: desc if sort_by provided. Returns: Dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ params = DelphiVideoPositionsChartsV3.RequestSchema().dump(locals()) return await self.get("video-positions/charts", params) async def get_regions( self, offset: int or None = None, limit: int or None = None, sort_by: str or None = None, sort_order: str or None = None, ) -> Dict[str, Any]: """Get youtube video streams with additional data. Args: offset: pagination offset. limit: pagination limit. sort_by: top level object field name which is used to sort the results by. sort_order: desc / asc, desc by default. Returns: dictionary with "count" and "items". Pay attention: @pagination decorator changes method's return type, wrapped method return list instead dict. """ params = DelphiRegionsParams().dump(locals()) return await self.get("regions", params) async def get_all_country_codes(self) -> List[str]: result = await self.get_regions() return [i["country_code"] for i in result["items"]] async def get_tracks(self, **params) -> List[dict]: """Get tracks data from Delphi API. Returns: Tracks data. """ result = await self.get("tracks", params) return result["items"] async def _get_tiktok_tracks_analytics( self, isrc_list: List[str], start_date: date, end_date: date, content_type_list: List[str] or None, country_code_list: List[str] or None, breakdowns_list: List[str] or None, metrics_list: List[str] or None = None, expands_list: List[str] or None = None, ) -> Dict[str, Any]: """Get tiktok tracks analytics data from Delphi API. Args: isrc_list: International Standard Recording Code (ISRC) number list. start_date: Get data from this date. end_date: Get data to this date. content_type_list: List of content types, e.g.: partner_uploaded, premium_ugc, etc. country_code_list: List of two letter country codes or worldwide. breakdowns_list: List of breakdowns for analytics data. metrics_list: List of fields in result. expands_list: 'related_isrcs' is the only available option. Returns: Tiktok analytics data. """ params = DelphiTiktokTracksAnalyticsV3RequestSchema().dump(locals()) return await self.get("tiktok/tracks/analytics", params) @request_per_item( "isrc_list", filter_kwargs=True, sum_func=lambda r, c, _ars, _kwargs: {_kwargs["isrc_list"][0]: c, **r}, result_cls=dict, ) async def get_tiktok_tracks_analytics_by_isrc( self, isrc_list: List[str], start_date: date, end_date: date, content_type_list: List[str] or None = None, country_code_list: List[str] or None = None, breakdowns_list: List[str] or None = None, metrics_list: List[str] or None = None, all_countries_totals: bool = True, expands_list: List[str] or None = None, ) -> Dict[str, Any]: """Get analytics data in format {item_isrc: analytics data}""" return await self.get_tiktok_tracks_analytics_bulk( isrc_list=isrc_list, start_date=start_date, end_date=end_date, content_type_list=content_type_list, country_code_list=country_code_list, breakdowns_list=breakdowns_list, metrics_list=metrics_list, all_countries_totals=all_countries_totals, expands_list=expands_list, ) @request_per_item( "isrc_list", sum_all_func=sum_tiktok_tracks_analytics, modify_items_func=request_related_isrc, filter_kwargs=True, ) async def get_tiktok_tracks_analytics( self, isrc_list: List[str], start_date: date, end_date: date, content_type_list: List[str] or None = None, country_code_list: List[str] or None = None, breakdowns_list: List[str] or None = None, metrics_list: List[str] or None = None, all_countries_totals: bool = True, expands_list: List[str] or None = None, ) -> Dict[str, Any]: """Get combined by isrc data.""" return await self.get_tiktok_tracks_analytics_bulk( isrc_list=isrc_list, start_date=start_date, end_date=end_date, content_type_list=content_type_list, country_code_list=country_code_list, breakdowns_list=breakdowns_list, metrics_list=metrics_list, all_countries_totals=all_countries_totals, expands_list=expands_list, ) async def get_tiktok_tracks_analytics_bulk( self, isrc_list: List[str], start_date: date, end_date: date, content_type_list: List[str] or None = None, country_code_list: List[str] or None = None, breakdowns_list: List[str] or None = None, metrics_list: List[str] or None = None, all_countries_totals: bool = True, expands_list: List[str] or None = None, ) -> Dict[str, Any]: """Get tiktok tracks analytics data with custom filters and totals logic. Args: isrc_list: International Standard Recording Code (ISRC) number list. start_date: Get data from this date. end_date: Get data to this date. content_type_list: List of content types, e.g.: partner_uploaded, premium_ugc, etc. country_code_list: List of two letter country codes or worldwide. breakdowns_list: List of breakdowns for analytics data. metrics_list: List of fields in result. all_countries_totals: All countries for breakdowns=country_totals. expands_list: List of expands. Returns: Tiktok analytics data. """ if not breakdowns_list: breakdowns_list = [TiktokTrackAnalyticsBreakdown.DAILY] has_totals = TiktokTrackAnalyticsBreakdown.TOTALS in breakdowns_list original_country_code_list = None if country_code_list is None else list(country_code_list) all_countries = (country_code_list and ALL in country_code_list) or ( not country_code_list and TiktokTrackAnalyticsBreakdown.COUNTRY_TOTALS in breakdowns_list and all_countries_totals ) if all_countries: country_code_list = await self.get_all_country_codes() original_breakdowns_list = list(breakdowns_list) if has_totals: breakdowns_list = update_breakdowns_for_totals(breakdowns_list) result = await self._get_tiktok_tracks_analytics( isrc_list, start_date, end_date, content_type_list, country_code_list, breakdowns_list, metrics_list, expands_list, ) result_breakdowns = result["breakdowns"] if has_totals: count_totals(result_breakdowns, original_breakdowns_list, original_country_code_list, all_countries) return result @pagination(config.DELPHI_CHARTS_PAGE_SIZE, response_items_node=None, response_count_node=None) async def get_shazam_charts_track_positions(self, **params) -> List[dict]: """Get shazam charts positions data from Delphi API. Returns: Charts positions data. """ result = await self.get("shazam/charts/track-positions", params) return result["items"] @prepare_list_arg("isrc") async def get_shazam_charts_track_positions_summary(self, **params) -> List[dict]: """Get shazam charts summary data from Delphi API. Returns: Charts summary data. """ result = await self.get("shazam/charts/track-positions/summary", params) return result["items"] async def get_shazam_cities(self, **params) -> List[dict]: """Get shazam cities from Delphi API. Returns: Shazam cities. """ result = await self.get("shazam/cities", params) return result["items"] @request_per_item("isrc_list", sum_all_func=sum_chart_analytics) async def get_charts_analytics( self, dsp: str, chart_id: str, start_date: date, end_date: date, metrics: List[str], group_by: str, track_id_list: Optional[List[str]] = None, isrc_list: Optional[List[str]] = None, ): params = DelphiChartsAnalytics.DelphiRequest().dump(locals()) result = await self._get(f"{DSP_URL_MAPPING[dsp]}/charts/analytics", params) return result["items"] async def get_tracks_charts( self, dsp: str, track_id_list: Optional[List[str]] = None, isrc_list: Optional[List[str]] = None, chart_date: Optional[date] = None, track_state_includes_list: Optional[List[str]] = None, metrics_dimension: Optional[List[str]] = None, chart_type_list: Optional[List[str]] = None, chart_breakdown_list: Optional[List[str]] = None, chart_country_code_list: Optional[List[str]] = None, chart_min_rank: Optional[int] = None, chart_max_rank: Optional[int] = None, min_track_position: Optional[int] = None, max_track_position: Optional[int] = None, ): # Two possible values. # Delphi API has no default value (returns all data when no filtering) and accepts only one value as a request # parameter, so the list of two values is equivalent of None. if chart_breakdown_list and len(chart_breakdown_list) > 1: chart_breakdown_list = None if chart_type_list and len(chart_type_list) > 1: chart_type_list = None params = DelphiTracksCharts.DelphiRequest().dump(locals()) result = await self._get(f"{DSP_URL_MAPPING[dsp]}/tracks/charts", params) return result["items"] async def get_playlists_followers( self, playlist_id: List[str], start_date: date, end_date: date, fill_skipped: bool = True ): params = DelphiPlaylistsFollowers.DelphiRequest().dump(locals()) result = await self._get(f"public/playlists/followers", params) if not fill_skipped: return result # handling delphi bug: if there is the first (or last) item with followers = null, it is excluded from # the response (for non first/last items Delphi returns null in the same case), # so we manually add skipped null(s) return { "count": result["count"], "items": [ fill_skipped_dates(item, items_key="followers", request_dates=(start_date, end_date)) for item in result["items"] ], } async def check_health(self): """Health check.""" result = await self.get("health") return result["status"] == "UP"