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)