I wrote this decorator to be used by two functions:
def next_page(func) -> List[Dict]: """ Decorator to re-run api call if number of objects returned is 100. SmartSurvey API can only return objects for 1 "page" at each call. One page stores 100 objects. By passing the argument page += 1 you can get the next 100 records till the call return less than 100. :param func: :return: list of dictionaries """ @functools.wraps(func) def wrapper(ASO, page, **kwargs): list_of_records = [] partial_list = func(ASO, page, **kwargs) list_of_records.extend(partial_list) while len(partial_list) == 100: page += 1 partial_list = func(ASO, page, **kwargs) list_of_records.extend(partial_list) return list_of_records return wrapper @next_page def _get_responses(ASO, page, survey_id, since) -> List[Dict]: """ Get responses from SmartSurvey API based on survey id. :param ASO: API object :param survey_id: INTEGER :param page: INTEGER :param since: INTEGER (unix epoch) :return: list of dictionaries """ responses = ASO.get_responses(page=page, survey_id=survey_id, since=since, page_size=100) return responses @next_page def _get_surveys(ASO, page) -> List[Dict]: """ Get surveys from SmartSurvey API. :param ASO: API object :param page: INTEGER :return: list of dictionaries """ surveys = ASO.get_surveys(page=page, page_size=100) return surveys
I keep getting the error:
Traceback (most recent call last): File "/Users/asportelli/gitRepos/smart-survey/scripts/api_to_s3_local.py", line 206, in <module> main() File "/Users/asportelli/gitRepos/smart-survey/scripts/api_to_s3_local.py", line 202, in main responses_to_s3(s3, API, survey_meta_data, _last_n_hours(1)) File "/Users/asportelli/gitRepos/smart-survey/scripts/api_to_s3_local.py", line 160, in responses_to_s3 since=since) TypeError: wrapper() got multiple values for argument 'page'
How can I pass the argument page recursively without having it multiple times?
Thanks!
EDIT:
functions are called as in:
def surveys_metadata(ASO) -> pd.DataFrame: """ get id and last updated ts of all surveys :param ASO: Api Scraper Object :param from_ts: unix epoch timestamp :return: pandas df """ dfs = [] surveys = _get_surveys(ASO, page=1) for survey in surveys: df = pd.DataFrame( {'id': [survey['id']], 'date_modified': [survey['date_modified']]}) dfs.append(df) output = pd.concat(dfs) logger.info(f"surveys_found: {len(output)}") return output def responses_to_s3(s3, ASO, df, since=_last_n_hours(24 * 1)): """ Write SmartSurvey responses to AWS S3 bucket. :param s3: boto3 client :param ASO: API object :param df: smart survey metadata (pd.DataFrame) :param since: INTEGER (unix epoch) :return: """ for survey_id in df['id']: responses = _get_responses(ASO, survey_id, page=1, since=since) logger.info(f"Extracted {len(responses)} from survey: {survey_id}") for response in responses: response_id = response['id'] response_ts = response['date_modified'] date = str(response['date_modified'])[:10] _dict_to_s3(s3, 'dundata-nonprod-lake', f'to_process/smart-survey/responses/{date}/' f'{survey_id}-{response_id}-{response_ts}.json', response)
Advertisement
Answer
Positional arguments in python must be passed in order of definition in the function prototype.
This line passes the value of survey_id
to the argument page
as it’s the second argument passed.
responses = _get_responses(ASO, survey_id, page=1, since=since)
Call the function as:
responses = _get_responses(ASO, 1, survey_id=survey_id, since=since)