from typing import List from apollo_utils.core.utils.result_types import DictResult from apollo_utils.job.clients.base.client import BaseApiClient from apollo_utils.job.clients.base.config import BaseConfig from apollo_utils.job.clients.utils.request_in_chunks.sync.request_in_chunks import request_in_chunks from charts_messages.logger import get_logger from charts_messages.config import CHUNK_WAIT from charts_messages.constants import Service, USER_DATA_HEADER, CHUNK_SIZE, PostFavoritesInclude, APP_NAME from charts_messages.utils import create_post_favorites_body class GateApiClientConfig(BaseConfig): service = Service.GATE_API logger = get_logger(APP_NAME) def __init__(self): self._load_attrs() super().__init__() class GateApiClient(GateApiClientConfig, BaseApiClient): @request_in_chunks( chunk_size=CHUNK_SIZE, items_kwarg="isrc_list", result_type=DictResult, wait=True, wait_time=CHUNK_WAIT ) def get_users(self, isrc_list: List[str], **params): """Get users information based on body passed filters Args: isrc_list: """ body = create_post_favorites_body( isrc=isrc_list, include=[ PostFavoritesInclude.SETTINGS.value, PostFavoritesInclude.DEVICES.value ] ) return self.send_request( "api/service/users/favorites/list/", method="POST", data=body, headers=USER_DATA_HEADER, get_json=True, **params )