From: Harmon Date: Wed, 10 Feb 2021 03:33:17 +0000 (-0600) Subject: Merge branch 'master' into video-upload X-Git-Url: https://vcs.fsf.org/?a=commitdiff_plain;h=7194ef8c0e11d418c527c4b73d88f931a00d42ff;p=tweepy.git Merge branch 'master' into video-upload --- 7194ef8c0e11d418c527c4b73d88f931a00d42ff diff --cc tweepy/api.py index 77c35da,846ed3d..503e76f --- a/tweepy/api.py +++ b/tweepy/api.py @@@ -75,438 -93,383 +93,484 @@@ class API f' It is currently a {type(self.parser)}.' ) - @property - def home_timeline(self): + self.session = requests.Session() + + def request( + self, method, endpoint, *args, endpoint_parameters=(), params=None, + headers=None, json_payload=None, parser=None, payload_list=False, - payload_type=None, post_data=None, require_auth=True, ++ payload_type=None, post_data=None, files=None, require_auth=True, + return_cursors=False, upload_api=False, use_cache=True, **kwargs + ): + # If authentication is required and no credentials + # are provided, throw an error. + if require_auth and not self.auth: + raise TweepError('Authentication required!') + + self.cached_result = False + + # Build the request URL + path = f'/1.1/{endpoint}.json' + if upload_api: + url = 'https://' + self.upload_host + path + else: + url = 'https://' + self.host + path + + if params is None: + params = {} + + for idx, arg in enumerate(args): + if arg is None: + continue + try: + params[endpoint_parameters[idx]] = str(arg) + except IndexError: + raise TweepError('Too many parameters supplied!') + + for k, arg in kwargs.items(): + if arg is None: + continue + if k in params: + raise TweepError(f'Multiple values for parameter {k} supplied!') + if k not in endpoint_parameters: + log.warning(f'Unexpected parameter: {k}') + params[k] = str(arg) + + log.debug("PARAMS: %r", params) + + # Query the cache if one is available + # and this request uses a GET method. + if use_cache and self.cache and method == 'GET': + cache_result = self.cache.get(f'{path}?{urlencode(params)}') + # if cache result found and not expired, return it + if cache_result: + # must restore api reference + if isinstance(cache_result, list): + for result in cache_result: + if isinstance(result, Model): + result._api = self + else: + if isinstance(cache_result, Model): + cache_result._api = self + self.cached_result = True + return cache_result + + # Monitoring rate limits + remaining_calls = None + reset_time = None + + if parser is None: + parser = self.parser + + try: + # Continue attempting request until successful + # or maximum number of retries is reached. + retries_performed = 0 + while retries_performed <= self.retry_count: + if (self.wait_on_rate_limit and reset_time is not None + and remaining_calls is not None + and remaining_calls < 1): + # Handle running out of API calls + sleep_time = reset_time - int(time.time()) + if sleep_time > 0: + log.warning(f"Rate limit reached. Sleeping for: {sleep_time}") + time.sleep(sleep_time + 1) # Sleep for extra sec + + # Apply authentication + auth = None + if self.auth: + auth = self.auth.apply_auth() + + # Execute request + try: + resp = self.session.request( + method, url, params=params, headers=headers, - data=post_data, json=json_payload, timeout=self.timeout, - auth=auth, proxies=self.proxy ++ data=post_data, files=files, json=json_payload, ++ timeout=self.timeout, auth=auth, proxies=self.proxy + ) + except Exception as e: + raise TweepError(f'Failed to send request: {e}').with_traceback(sys.exc_info()[2]) + + if 200 <= resp.status_code < 300: + break + + rem_calls = resp.headers.get('x-rate-limit-remaining') + if rem_calls is not None: + remaining_calls = int(rem_calls) + elif remaining_calls is not None: + remaining_calls -= 1 + + reset_time = resp.headers.get('x-rate-limit-reset') + if reset_time is not None: + reset_time = int(reset_time) + + retry_delay = self.retry_delay + if resp.status_code in (420, 429) and self.wait_on_rate_limit: + if remaining_calls == 0: + # If ran out of calls before waiting switching retry last call + continue + if 'retry-after' in resp.headers: + retry_delay = float(resp.headers['retry-after']) + elif self.retry_errors and resp.status_code not in self.retry_errors: + # Exit request loop if non-retry error code + break + + # Sleep before retrying request again + time.sleep(retry_delay) + retries_performed += 1 + + # If an error was returned, throw an exception + self.last_response = resp + if resp.status_code and not 200 <= resp.status_code < 300: + try: + error_msg, api_error_code = parser.parse_error(resp.text) + except Exception: + error_msg = f"Twitter error response: status code = {resp.status_code}" + api_error_code = None + + if is_rate_limit_error_message(error_msg): + raise RateLimitError(error_msg, resp) + else: + raise TweepError(error_msg, resp, api_code=api_error_code) + + # Parse the response payload + return_cursors = return_cursors or 'cursor' in params or 'next' in params + result = parser.parse( + resp.text, api=self, payload_list=payload_list, + payload_type=payload_type, return_cursors=return_cursors + ) + + # Store result into cache if one is available. + if use_cache and self.cache and method == 'GET' and result: + self.cache.store(f'{path}?{urlencode(params)}', result) + + return result + finally: + self.session.close() + + @pagination(mode='id') + @payload('status', list=True) + def home_timeline(self, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-home_timeline - :allowed_param: 'count', 'since_id', 'max_id', 'trim_user', - 'exclude_replies', 'include_entities' """ - return bind_api( - api=self, - path='/statuses/home_timeline.json', - payload_type='status', payload_list=True, - allowed_param=['count', 'since_id', 'max_id', 'trim_user', - 'exclude_replies', 'include_entities'], - require_auth=True + return self.request( + 'GET', 'statuses/home_timeline', endpoint_parameters=( + 'count', 'since_id', 'max_id', 'trim_user', 'exclude_replies', + 'include_entities' + ), **kwargs ) - def statuses_lookup(self, id_, *args, **kwargs): + @payload('status', list=True) + def statuses_lookup(self, id, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-lookup - :allowed_param: 'id', 'include_entities', 'trim_user', 'map', - 'include_ext_alt_text', 'include_card_uri' - """ - if 'map_' in kwargs: - kwargs['map'] = kwargs.pop('map_') - - return bind_api( - api=self, - path='/statuses/lookup.json', - payload_type='status', payload_list=True, - allowed_param=['id', 'include_entities', 'trim_user', 'map', - 'include_ext_alt_text', 'include_card_uri'], - require_auth=True - )(list_to_csv(id_), *args, **kwargs) - - @property - def user_timeline(self): - """ :reference: https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-user_timeline - :allowed_param: 'user_id', 'screen_name', 'since_id', 'count', - 'max_id', 'trim_user', 'exclude_replies', - 'include_rts' """ - return bind_api( - api=self, - path='/statuses/user_timeline.json', - payload_type='status', payload_list=True, - allowed_param=['user_id', 'screen_name', 'since_id', 'count', - 'max_id', 'trim_user', 'exclude_replies', - 'include_rts'] + return self.request( + 'GET', 'statuses/lookup', list_to_csv(id), endpoint_parameters=( + 'id', 'include_entities', 'trim_user', 'map', + 'include_ext_alt_text', 'include_card_uri' + ), **kwargs ) - @property - def mentions_timeline(self): - """ :reference: https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-mentions_timeline - :allowed_param: 'since_id', 'max_id', 'count' + @pagination(mode='id') + @payload('status', list=True) + def user_timeline(self, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-user_timeline """ - return bind_api( - api=self, - path='/statuses/mentions_timeline.json', - payload_type='status', payload_list=True, - allowed_param=['since_id', 'max_id', 'count'], - require_auth=True + return self.request( + 'GET', 'statuses/user_timeline', endpoint_parameters=( + 'user_id', 'screen_name', 'since_id', 'count', 'max_id', + 'trim_user', 'exclude_replies', 'include_rts' + ), **kwargs ) - @property - def related_results(self): - """ :reference: https://dev.twitter.com/docs/api/1.1/get/related_results/show/%3id.format - :allowed_param: 'id' + @pagination(mode='id') + @payload('status', list=True) + def mentions_timeline(self, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-mentions_timeline """ - return bind_api( - api=self, - path='/related_results/show/{id}.json', - payload_type='relation', payload_list=True, - allowed_param=['id'], - require_auth=False + return self.request( + 'GET', 'statuses/mentions_timeline', endpoint_parameters=( + 'count', 'since_id', 'max_id', 'trim_user', 'include_entities' + ), **kwargs ) - @property - def retweets_of_me(self): + @pagination(mode='id') + @payload('status', list=True) + def retweets_of_me(self, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-retweets_of_me - :allowed_param: 'since_id', 'max_id', 'count' """ - return bind_api( - api=self, - path='/statuses/retweets_of_me.json', - payload_type='status', payload_list=True, - allowed_param=['since_id', 'max_id', 'count'], - require_auth=True + return self.request( + 'GET', 'statuses/retweets_of_me', endpoint_parameters=( + 'count', 'since_id', 'max_id', 'trim_user', 'include_entities', + 'include_user_entities' + ), **kwargs ) - @property - def get_status(self): + @payload('status') + def get_status(self, id, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-show-id - :allowed_param: 'id', 'trim_user', 'include_my_retweet', - 'include_entities', 'include_ext_alt_text', - 'include_card_uri' """ - return bind_api( - api=self, - path='/statuses/show.json', - payload_type='status', - allowed_param=['id', 'trim_user', 'include_my_retweet', - 'include_entities', 'include_ext_alt_text', - 'include_card_uri'] + return self.request( + 'GET', 'statuses/show', id, endpoint_parameters=( + 'id', 'trim_user', 'include_my_retweet', 'include_entities', + 'include_ext_alt_text', 'include_card_uri' + ), **kwargs ) - def update_status(self, *args, **kwargs): + @payload('status') + def update_status(self, status, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update - :allowed_param: 'status', 'in_reply_to_status_id', - 'auto_populate_reply_metadata', - 'exclude_reply_user_ids', 'attachment_url', - 'media_ids', 'possibly_sensitive', 'lat', 'long', - 'place_id', 'display_coordinates', 'trim_user', - 'enable_dmcommands', 'fail_dmcommands', 'card_uri' """ if 'media_ids' in kwargs: kwargs['media_ids'] = list_to_csv(kwargs['media_ids']) - return bind_api( - api=self, - path='/statuses/update.json', - method='POST', - payload_type='status', - allowed_param=['status', 'in_reply_to_status_id', - 'auto_populate_reply_metadata', - 'exclude_reply_user_ids', 'attachment_url', - 'media_ids', 'possibly_sensitive', 'lat', 'long', - 'place_id', 'display_coordinates', 'trim_user', - 'enable_dmcommands', 'fail_dmcommands', - 'card_uri'], - require_auth=True - )(*args, **kwargs) + return self.request( + 'POST', 'statuses/update', status, endpoint_parameters=( + 'status', 'in_reply_to_status_id', + 'auto_populate_reply_metadata', 'exclude_reply_user_ids', + 'attachment_url', 'media_ids', 'possibly_sensitive', 'lat', + 'long', 'place_id', 'display_coordinates', 'trim_user', + 'enable_dmcommands', 'fail_dmcommands', 'card_uri' + ), **kwargs + ) - @payload('media') - def media_upload(self, filename, *args, **kwargs): + def media_upload(self, filename, file=None, chunked=False, + *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/media/upload-media/api-reference/post-media-upload - :allowed_param: """ - f = kwargs.pop('file', None) - h = None - if f is not None: - location = f.tell() - h = f.read(32) - f.seek(location) - file_type = imghdr.what(filename, h=h) or mimetypes.guess_type(filename)[0] - if file_type == 'gif': - max_size = 14649 + if file is not None: + location = file.tell() + h = file.read(32) + file.seek(location) + file_type = imghdr.what(filename, h=h) + if file_type is not None: + file_type = 'image/' + file_type + else: + file_type = mimetypes.guess_type(filename)[0] + + if chunked or file_type.startswith('video/'): + return self.chunked_upload(filename, file=file, + file_type=file_type, *args, **kwargs) else: - max_size = 4883 + return self.simple_upload(filename, file=file, *args, **kwargs) + ++ @payload('media') + def simple_upload(self, filename, file=None, media_category=None, + *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/media/upload-media/api-reference/post-media-upload - :allowed_param: + """ + if file is not None: + files = {'media': (filename, file)} + else: + files = {'media': open(filename, 'rb')} + post_data = {'media_category': media_category} - return bind_api( - api=self, - path='/media/upload.json', - method='POST', - payload_type='media', - require_auth=True, - upload_api=True - )(*args, post_data=post_data, files=files, **kwargs) ++ return self.request( ++ 'POST', 'media/upload', *args, post_data=post_data, files=files, ++ upload_api=True, **kwargs ++ ) + + def chunked_upload(self, filename, file=None, file_type=None, + media_category=None, *args, **kwargs): + """ :reference https://developer.twitter.com/en/docs/media/upload-media/uploading-media/chunked-media-upload - :allowed_param: + """ + fp = file or open(filename, 'rb') - headers, post_data = API._pack_image(filename, max_size, - form_field='media', f=f, - file_type=file_type) - kwargs.update({'headers': headers, 'post_data': post_data}) + start = fp.tell() + fp.seek(0, 2) # Seek to end of file + file_size = fp.tell() - start + fp.seek(start) + media_id = self.chunked_upload_init( + file_size, file_type, media_category, *args, **kwargs + ).media_id + + min_chunk_size, remainder = divmod(file_size, 1000) + min_chunk_size += bool(remainder) + + # Use 1 MiB as default chunk size + chunk_size = kwargs.pop('chunk_size', 1024 * 1024) + # Max chunk size is 5 MiB + chunk_size = max(min(chunk_size, 5 * 1024 * 1024), min_chunk_size) + + segments, remainder = divmod(file_size, chunk_size) + segments += bool(remainder) + + for segment_index in range(segments): + # The APPEND command returns an empty response body + self.chunked_upload_append( + media_id, (filename, fp.read(chunk_size)), segment_index, + *args, **kwargs + ) + + fp.close() + # The FINALIZE command returns media information + return self.chunked_upload_finalize(media_id, *args, **kwargs) + ++ @payload('media') + def chunked_upload_init(self, total_bytes, media_type, media_category=None, + additional_owners=None, *args, **kwargs): + """ :reference https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload-init - :allowed_param: + """ + headers = {'Content-Type': 'application/x-www-form-urlencoded'} + post_data = { + 'command': 'INIT', + 'total_bytes': total_bytes, + 'media_type': media_type, + 'media_category': media_category + } + if additional_owners is not None: + post_data["additional_owners"] = list_to_csv(additional_owners) - return bind_api( - api=self, - path='/media/upload.json', - method='POST', - payload_type='media', - require_auth=True, - upload_api=True - )(*args, headers=headers, post_data=post_data, **kwargs) ++ return self.request( ++ 'POST', 'media/upload', *args, headers=headers, ++ post_data=post_data, upload_api=True, **kwargs ++ ) + + def chunked_upload_append(self, media_id, media, segment_index, + *args, **kwargs): + """ :reference https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload-append - :allowed_param: + """ + post_data = { + 'command': 'APPEND', + 'media_id': media_id, + 'segment_index': segment_index + } + files = {'media': media} - return bind_api( - api=self, - path='/media/upload.json', - method='POST', - require_auth=True, - upload_api=True - )(*args, post_data=post_data, files=files, **kwargs) + return self.request( - 'POST', 'media/upload', *args, - endpoint_parameters=(), ++ 'POST', 'media/upload', *args, post_data=post_data, files=files, + upload_api=True, **kwargs + ) ++ @payload('media') + def chunked_upload_finalize(self, media_id, *args, **kwargs): + """ :reference https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload-finalize - :allowed_param: + """ + headers = {'Content-Type': 'application/x-www-form-urlencoded'} + post_data = { + 'command': 'FINALIZE', + 'media_id': media_id + } - return bind_api( - api=self, - path='/media/upload.json', - method='POST', - payload_type='media', - require_auth=True, - upload_api=True - )(*args, headers=headers, post_data=post_data, **kwargs) ++ return self.request( ++ 'POST', 'media/upload', *args, headers=headers, ++ post_data=post_data, upload_api=True, **kwargs ++ ) + def create_media_metadata(self, media_id, alt_text, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/media/upload-media/api-reference/post-media-metadata-create - :allowed_param: """ kwargs['json_payload'] = { 'media_id': media_id, 'alt_text': {'text': alt_text} } - return bind_api( - api=self, - path='/media/metadata/create.json', - method='POST', - require_auth=True, - upload_api=True - )(*args, **kwargs) + return self.request( + 'POST', 'media/metadata/create', *args, - endpoint_parameters=(), + upload_api=True, **kwargs + ) + @payload('status') - def update_with_media(self, filename, *args, **kwargs): + def update_with_media(self, filename, file=None, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update_with_media - :allowed_param: 'status', 'possibly_sensitive', - 'in_reply_to_status_id', - 'in_reply_to_status_id_str', - 'auto_populate_reply_metadata', 'lat', 'long', - 'place_id', 'display_coordinates' """ - f = kwargs.pop('file', None) - headers, post_data = API._pack_image(filename, 3072, - form_field='media[]', f=f) - kwargs.update({'headers': headers, 'post_data': post_data}) - + if file is not None: + files = {'media[]': (filename, file)} + else: + files = {'media[]': open(filename, 'rb')} - return bind_api( - api=self, - path='/statuses/update_with_media.json', - method='POST', - payload_type='status', - allowed_param=['status', 'possibly_sensitive', - 'in_reply_to_status_id', - 'in_reply_to_status_id_str', - 'auto_populate_reply_metadata', 'lat', 'long', - 'place_id', 'display_coordinates'], - require_auth=True - )(*args, files=files, **kwargs) + return self.request( - 'POST', 'statuses/update_with_media', *args, endpoint_parameters=( ++ 'POST', 'statuses/update_with_media', *args, endpoint_parameters=( + 'status', 'possibly_sensitive', 'in_reply_to_status_id', + 'in_reply_to_status_id_str', 'auto_populate_reply_metadata', + 'lat', 'long', 'place_id', 'display_coordinates' - ), **kwargs ++ ), files=files, **kwargs ++ ) + ++ @payload('media') + def get_media_upload_status(self, media_id, *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/get-media-upload-status - :allowed_param: 'media_id' - """ - return bind_api( - api=self, - path='/media/upload.json', - payload_type='media', - allowed_param=['media_id'], - require_auth=True, - upload_api=True - )(*args, command='STATUS', media_id=media_id, **kwargs) - - @property - def destroy_status(self): ++ """ ++ return self.request( ++ 'GET', 'media/upload', *args, endpoint_parameters=( ++ 'command', 'media_id' ++ ), command='STATUS', media_id=media_id, upload_api=True, **kwargs + ) + + @payload('status') + def destroy_status(self, status_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-destroy-id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/statuses/destroy/{id}.json', - method='POST', - payload_type='status', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', f'statuses/destroy/{status_id}', *args, **kwargs ) - @property - def retweet(self): + @payload('status') + def retweet(self, status_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-retweet-id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/statuses/retweet/{id}.json', - method='POST', - payload_type='status', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', f'statuses/retweet/{status_id}', *args, **kwargs ) - @property - def unretweet(self): + @payload('status') + def unretweet(self, status_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-unretweet-id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/statuses/unretweet/{id}.json', - method='POST', - payload_type='status', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', f'statuses/unretweet/{status_id}', *args, **kwargs ) - @property - def retweets(self): + @payload('status', list=True) + def retweets(self, status_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-retweets-id - :allowed_param: 'id', 'count' """ - return bind_api( - api=self, - path='/statuses/retweets/{id}.json', - payload_type='status', payload_list=True, - allowed_param=['id', 'count'], - require_auth=True + return self.request( + 'GET', f'statuses/retweets/{status_id}', *args, + endpoint_parameters=( + 'count', + ), **kwargs ) - @property - def retweeters(self): + @pagination(mode='cursor') + @payload('ids') + def retweeters(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-retweeters-ids - :allowed_param: 'id', 'cursor', 'stringify_ids """ - return bind_api( - api=self, - path='/statuses/retweeters/ids.json', - payload_type='ids', - allowed_param=['id', 'cursor', 'stringify_ids'] + return self.request( + 'GET', 'statuses/retweeters/ids', *args, endpoint_parameters=( + 'id', 'cursor', 'stringify_ids' + ), **kwargs ) - @property - def get_user(self): + @payload('user') + def get_user(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-users-show - :allowed_param: 'id', 'user_id', 'screen_name' """ - return bind_api( - api=self, - path='/users/show.json', - payload_type='user', - allowed_param=['id', 'user_id', 'screen_name'] + return self.request( + 'GET', 'users/show', *args, endpoint_parameters=( + 'id', 'user_id', 'screen_name' + ), **kwargs ) - @property - def get_oembed(self): + @payload('json') + def get_oembed(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-oembed - :allowed_param: 'url', 'maxwidth', 'hide_media', 'hide_thread', - 'omit_script', 'align', 'related', 'lang', 'theme', - 'link_color', 'widget_type', 'dnt' """ - return bind_api( - api=self, - path='/statuses/oembed.json', - payload_type='json', - allowed_param=['url', 'maxwidth', 'hide_media', 'hide_thread', - 'omit_script', 'align', 'related', 'lang', 'theme', - 'link_color', 'widget_type', 'dnt'] + return self.request( + 'GET', 'statuses/oembed', *args, endpoint_parameters=( + 'url', 'maxwidth', 'hide_media', 'hide_thread', 'omit_script', + 'align', 'related', 'lang', 'theme', 'link_color', + 'widget_type', 'dnt' + ), require_auth=False, **kwargs ) + @payload('user', list=True) def lookup_users(self, user_ids=None, screen_names=None, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-users-lookup - allowed_param= 'user_id', 'screen_name', 'include_entities', - 'tweet_mode' - """ - return bind_api( - api=self, - path='/users/lookup.json', - payload_type='user', payload_list=True, - method='POST', - allowed_param=['user_id', 'screen_name', 'include_entities', - 'tweet_mode'] - )(list_to_csv(user_ids), list_to_csv(screen_names), *args, **kwargs) + """ + return self.request( + 'POST', 'users/lookup', list_to_csv(user_ids), + list_to_csv(screen_names), *args, endpoint_parameters=( + 'user_id', 'screen_name', 'include_entities', 'tweet_mode' + ), **kwargs + ) def me(self): """ Get the authenticated user """ @@@ -780,698 -695,581 +796,525 @@@ return False raise - @property - def rate_limit_status(self): + @payload('json') + def rate_limit_status(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/developer-utilities/rate-limit-status/api-reference/get-application-rate_limit_status - :allowed_param: 'resources' """ - return bind_api( - api=self, - path='/application/rate_limit_status.json', - payload_type='json', - allowed_param=['resources'], - use_cache=False + return self.request( + 'GET', 'application/rate_limit_status', *args, + endpoint_parameters=( + 'resources', + ), use_cache=False, **kwargs ) + @payload('user') - def update_profile_image(self, filename, file_=None, **kwargs): + def update_profile_image(self, filename, file=None, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-update_profile_image - :allowed_param: 'include_entities', 'skip_status' """ - headers, post_data = API._pack_image(filename, 700, f=file_) + if file is not None: + files = {'image': (filename, file)} + else: + files = {'image': open(filename, 'rb')} - return bind_api( - api=self, - path='/account/update_profile_image.json', - method='POST', - payload_type='user', - allowed_param=['include_entities', 'skip_status'], - require_auth=True - )(*args, files=files, **kwargs) + return self.request( - 'POST', 'account/update_profile_image', *args, ++ 'POST', 'account/update_profile_image', *args, + endpoint_parameters=( + 'include_entities', 'skip_status' - ), post_data=post_data, headers=headers, **kwargs ++ ), files=files, **kwargs + ) - def update_profile_banner(self, filename, **kwargs): + def update_profile_banner(self, filename, file=None, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-update_profile_banner - :allowed_param: 'width', 'height', 'offset_left', 'offset_right' """ - f = kwargs.pop('file', None) - headers, post_data = API._pack_image(filename, 700, - form_field='banner', f=f) + if file is not None: + files = {'banner': (filename, file)} + else: + files = {'banner': open(filename, 'rb')} - return bind_api( - api=self, - path='/account/update_profile_banner.json', - method='POST', - allowed_param=['width', 'height', 'offset_left', 'offset_right'], - require_auth=True - )(*args, files=files, **kwargs) - - @property - def update_profile(self): + return self.request( - 'POST', 'account/update_profile_banner', endpoint_parameters=( ++ 'POST', 'account/update_profile_banner', *args, ++ endpoint_parameters=( + 'width', 'height', 'offset_left', 'offset_right' - ), post_data=post_data, headers=headers, **kwargs ++ ), files=files, **kwargs + ) + + @payload('user') + def update_profile(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-update_profile - :allowed_param: 'name', 'url', 'location', 'description', - 'profile_link_color' """ - return bind_api( - api=self, - path='/account/update_profile.json', - method='POST', - payload_type='user', - allowed_param=['name', 'url', 'location', 'description', - 'profile_link_color'], - require_auth=True + return self.request( + 'POST', 'account/update_profile', *args, endpoint_parameters=( + 'name', 'url', 'location', 'description', 'profile_link_color' + ), **kwargs ) - @property - def favorites(self): + @pagination(mode='id') + @payload('status', list=True) + def favorites(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-favorites-list - :allowed_param: 'screen_name', 'user_id', 'max_id', 'count', - 'since_id', 'max_id' """ - return bind_api( - api=self, - path='/favorites/list.json', - payload_type='status', payload_list=True, - allowed_param=['screen_name', 'user_id', 'max_id', 'count', - 'since_id', 'max_id'] + return self.request( + 'GET', 'favorites/list', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'max_id', 'count', 'since_id' + ), **kwargs ) - @property - def create_favorite(self): + @payload('status') + def create_favorite(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-favorites-create - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/favorites/create.json', - method='POST', - payload_type='status', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', 'favorites/create', *args, endpoint_parameters=( + 'id', + ), **kwargs ) - @property - def destroy_favorite(self): + @payload('status') + def destroy_favorite(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-favorites-destroy - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/favorites/destroy.json', - method='POST', - payload_type='status', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', 'favorites/destroy', *args, endpoint_parameters=( + 'id', + ), **kwargs ) - @property - def create_block(self): + @payload('user') + def create_block(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/post-blocks-create - :allowed_param: 'id', 'user_id', 'screen_name' """ - return bind_api( - api=self, - path='/blocks/create.json', - method='POST', - payload_type='user', - allowed_param=['id', 'user_id', 'screen_name'], - require_auth=True + return self.request( + 'POST', 'blocks/create', *args, endpoint_parameters=( + 'id', 'user_id', 'screen_name' + ), **kwargs ) - @property - def destroy_block(self): + @payload('user') + def destroy_block(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/post-blocks-destroy - :allowed_param: 'id', 'user_id', 'screen_name' """ - return bind_api( - api=self, - path='/blocks/destroy.json', - method='POST', - payload_type='user', - allowed_param=['id', 'user_id', 'screen_name'], - require_auth=True + return self.request( + 'POST', 'blocks/destroy', *args, endpoint_parameters=( + 'id', 'user_id', 'screen_name' + ), **kwargs ) - @property - def mutes_ids(self): + @pagination(mode='cursor') + @payload('ids') + def mutes_ids(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/get-mutes-users-ids - :allowed_param: 'cursor' """ - return bind_api( - api=self, - path='/mutes/users/ids.json', - payload_type='ids', - allowed_param=['cursor'], - require_auth=True + return self.request( + 'GET', 'mutes/users/ids', *args, endpoint_parameters=( + 'cursor', + ), **kwargs ) - @property - def mutes(self): + @pagination(mode='cursor') + @payload('user', list=True) + def mutes(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/get-mutes-users-list - :allowed_param: 'cursor', 'include_entities', 'skip_status' """ - return bind_api( - api=self, - path='/mutes/users/list.json', - payload_type='user', payload_list=True, - allowed_param=['cursor', 'include_entities', 'skip_status'], - required_auth=True + return self.request( + 'GET', 'mutes/users/list', *args, endpoint_parameters=( + 'cursor', 'include_entities', 'skip_status' + ), **kwargs ) - @property - def create_mute(self): + @payload('user') + def create_mute(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/post-mutes-users-create - :allowed_param: 'id', 'user_id', 'screen_name' """ - return bind_api( - api=self, - path='/mutes/users/create.json', - method='POST', - payload_type='user', - allowed_param=['id', 'user_id', 'screen_name'], - require_auth=True + return self.request( + 'POST', 'mutes/users/create', *args, endpoint_parameters=( + 'id', 'user_id', 'screen_name' + ), **kwargs ) - @property - def destroy_mute(self): + @payload('user') + def destroy_mute(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/post-mutes-users-destroy - :allowed_param: 'id', 'user_id', 'screen_name' """ - return bind_api( - api=self, - path='/mutes/users/destroy.json', - method='POST', - payload_type='user', - allowed_param=['id', 'user_id', 'screen_name'], - require_auth=True + return self.request( + 'POST', 'mutes/users/destroy', *args, endpoint_parameters=( + 'id', 'user_id', 'screen_name' + ), **kwargs ) - @property - def blocks(self): + @pagination(mode='cursor') + @payload('user', list=True) + def blocks(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/get-blocks-list - :allowed_param: 'cursor' """ - return bind_api( - api=self, - path='/blocks/list.json', - payload_type='user', payload_list=True, - allowed_param=['cursor'], - require_auth=True + return self.request( + 'GET', 'blocks/list', *args, endpoint_parameters=( + 'cursor', + ), **kwargs ) - @property - def blocks_ids(self): + @pagination(mode='cursor') + @payload('ids') + def blocks_ids(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/get-blocks-ids - :allowed_param: 'cursor' """ - return bind_api( - api=self, - path='/blocks/ids.json', - payload_type='ids', - allowed_param=['cursor'], - require_auth=True + return self.request( + 'GET', 'blocks/ids', *args, endpoint_parameters=( + 'cursor', + ), **kwargs ) - @property - def report_spam(self): + @payload('user') + def report_spam(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/mute-block-report-users/api-reference/post-users-report_spam - :allowed_param: 'user_id', 'screen_name', 'perform_block' """ - return bind_api( - api=self, - path='/users/report_spam.json', - method='POST', - payload_type='user', - allowed_param=['user_id', 'screen_name', 'perform_block'], - require_auth=True + return self.request( + 'POST', 'users/report_spam', *args, endpoint_parameters=( + 'user_id', 'screen_name', 'perform_block' + ), **kwargs ) - @property - def saved_searches(self): - """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/get-saved_searches-list """ - return bind_api( - api=self, - path='/saved_searches/list.json', - payload_type='saved_search', payload_list=True, - require_auth=True - ) + @payload('saved_search', list=True) + def saved_searches(self, *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/get-saved_searches-list + """ + return self.request('GET', 'saved_searches/list', *args, **kwargs) - @property - def get_saved_search(self): + @payload('saved_search') + def get_saved_search(self, saved_search_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/get-saved_searches-show-id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/saved_searches/show/{id}.json', - payload_type='saved_search', - allowed_param=['id'], - require_auth=True + return self.request( + 'GET', f'saved_searches/show/{saved_search_id}', *args, **kwargs ) - @property - def create_saved_search(self): + @payload('saved_search') + def create_saved_search(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-saved_searches-create - :allowed_param: 'query' """ - return bind_api( - api=self, - path='/saved_searches/create.json', - method='POST', - payload_type='saved_search', - allowed_param=['query'], - require_auth=True + return self.request( + 'POST', 'saved_searches/create', *args, endpoint_parameters=( + 'query', + ), **kwargs ) - @property - def destroy_saved_search(self): + @payload('saved_search') + def destroy_saved_search(self, saved_search_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-saved_searches-destroy-id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/saved_searches/destroy/{id}.json', - method='POST', - payload_type='saved_search', - allowed_param=['id'], - require_auth=True + return self.request( + 'POST', f'saved_searches/destroy/{saved_search_id}', *args, + **kwargs ) - @property - def create_list(self): + @payload('list') + def create_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-create - :allowed_param: 'name', 'mode', 'description' """ - return bind_api( - api=self, - path='/lists/create.json', - method='POST', - payload_type='list', - allowed_param=['name', 'mode', 'description'], - require_auth=True + return self.request( + 'POST', 'lists/create', *args, endpoint_parameters=( + 'name', 'mode', 'description' + ), **kwargs ) - @property - def destroy_list(self): + @payload('list') + def destroy_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-destroy - :allowed_param: 'owner_screen_name', 'owner_id', 'list_id', 'slug' """ - return bind_api( - api=self, - path='/lists/destroy.json', - method='POST', - payload_type='list', - allowed_param=['owner_screen_name', 'owner_id', 'list_id', 'slug'], - require_auth=True + return self.request( + 'POST', 'lists/destroy', *args, endpoint_parameters=( + 'owner_screen_name', 'owner_id', 'list_id', 'slug' + ), **kwargs ) - @property - def update_list(self): + @payload('list') + def update_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-update - :allowed_param: 'list_id', 'slug', 'name', 'mode', 'description', - 'owner_screen_name', 'owner_id' """ - return bind_api( - api=self, - path='/lists/update.json', - method='POST', - payload_type='list', - allowed_param=['list_id', 'slug', 'name', 'mode', 'description', - 'owner_screen_name', 'owner_id'], - require_auth=True + return self.request( + 'POST', 'lists/update', *args, endpoint_parameters=( + 'list_id', 'slug', 'name', 'mode', 'description', + 'owner_screen_name', 'owner_id' + ), **kwargs ) - @property - def lists_all(self): + @payload('list', list=True) + def lists_all(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-list - :allowed_param: 'screen_name', 'user_id', 'reverse' """ - return bind_api( - api=self, - path='/lists/list.json', - payload_type='list', payload_list=True, - allowed_param=['screen_name', 'user_id', 'reverse'], - require_auth=True + return self.request( + 'GET', 'lists/list', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'reverse' + ), **kwargs ) - @property - def lists_memberships(self): + @pagination(mode='cursor') + @payload('list', list=True) + def lists_memberships(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-memberships - :allowed_param: 'screen_name', 'user_id', 'filter_to_owned_lists', - 'cursor', 'count' """ - return bind_api( - api=self, - path='/lists/memberships.json', - payload_type='list', payload_list=True, - allowed_param=['screen_name', 'user_id', 'filter_to_owned_lists', - 'cursor', 'count'], - require_auth=True + return self.request( + 'GET', 'lists/memberships', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'filter_to_owned_lists', 'cursor', + 'count' + ), **kwargs ) - @property - def lists_ownerships(self): + @pagination(mode='cursor') + @payload('list', list=True) + def lists_ownerships(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-ownerships - :allowed_param: 'user_id', 'screen_name', 'count', 'cursor' """ - return bind_api( - api=self, - path='/lists/ownerships.json', - payload_type='list', payload_list=True, - allowed_param=['user_id', 'screen_name', 'count', 'cursor'], - require_auth=True + return self.request( + 'GET', 'lists/ownerships', *args, endpoint_parameters=( + 'user_id', 'screen_name', 'count', 'cursor' + ), **kwargs ) - @property - def lists_subscriptions(self): + @pagination(mode='cursor') + @payload('list', list=True) + def lists_subscriptions(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-subscriptions - :allowed_param: 'screen_name', 'user_id', 'cursor', 'count' """ - return bind_api( - api=self, - path='/lists/subscriptions.json', - payload_type='list', payload_list=True, - allowed_param=['screen_name', 'user_id', 'cursor', 'count'], - require_auth=True + return self.request( + 'GET', 'lists/subscriptions', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'cursor', 'count' + ), **kwargs ) - @property - def list_timeline(self): + @pagination(mode='id') + @payload('status', list=True) + def list_timeline(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-statuses - :allowed_param: 'owner_screen_name', 'slug', 'owner_id', 'list_id', - 'since_id', 'max_id', 'count', 'include_entities', - 'include_rts' """ - return bind_api( - api=self, - path='/lists/statuses.json', - payload_type='status', payload_list=True, - allowed_param=['owner_screen_name', 'slug', 'owner_id', 'list_id', - 'since_id', 'max_id', 'count', 'include_entities', - 'include_rts'] + return self.request( + 'GET', 'lists/statuses', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'owner_id', 'list_id', 'since_id', + 'max_id', 'count', 'include_entities', 'include_rts' + ), **kwargs ) - @property - def get_list(self): + @payload('list') + def get_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-show - :allowed_param: 'owner_screen_name', 'owner_id', 'slug', 'list_id' """ - return bind_api( - api=self, - path='/lists/show.json', - payload_type='list', - allowed_param=['owner_screen_name', 'owner_id', 'slug', 'list_id'] + return self.request( + 'GET', 'lists/show', *args, endpoint_parameters=( + 'owner_screen_name', 'owner_id', 'slug', 'list_id' + ), **kwargs ) - @property - def add_list_member(self): + @payload('list') + def add_list_member(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-members-create - :allowed_param: 'screen_name', 'user_id', 'owner_screen_name', - 'owner_id', 'slug', 'list_id' """ - return bind_api( - api=self, - path='/lists/members/create.json', - method='POST', - payload_type='list', - allowed_param=['screen_name', 'user_id', 'owner_screen_name', - 'owner_id', 'slug', 'list_id'], - require_auth=True + return self.request( + 'POST', 'lists/members/create', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'owner_screen_name', 'owner_id', + 'slug', 'list_id' + ), **kwargs ) - @property - def remove_list_member(self): + @payload('list') + def remove_list_member(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-members-destroy - :allowed_param: 'screen_name', 'user_id', 'owner_screen_name', - 'owner_id', 'slug', 'list_id' """ - return bind_api( - api=self, - path='/lists/members/destroy.json', - method='POST', - payload_type='list', - allowed_param=['screen_name', 'user_id', 'owner_screen_name', - 'owner_id', 'slug', 'list_id'], - require_auth=True + return self.request( + 'POST', 'lists/members/destroy', *args, endpoint_parameters=( + 'screen_name', 'user_id', 'owner_screen_name', 'owner_id', + 'slug', 'list_id' + ), **kwargs ) + @payload('list') def add_list_members(self, screen_name=None, user_id=None, slug=None, - list_id=None, owner_id=None, owner_screen_name=None): + list_id=None, owner_id=None, owner_screen_name=None, + **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-members-create_all - :allowed_param: 'screen_name', 'user_id', 'slug', 'list_id', - 'owner_id', 'owner_screen_name' - """ - return bind_api( - api=self, - path='/lists/members/create_all.json', - method='POST', - payload_type='list', - allowed_param=['screen_name', 'user_id', 'slug', 'list_id', - 'owner_id', 'owner_screen_name'], - require_auth=True - )(list_to_csv(screen_name), list_to_csv(user_id), slug, list_id, - owner_id, owner_screen_name) + """ + return self.request( + 'POST', 'lists/members/create_all', list_to_csv(screen_name), + list_to_csv(user_id), slug, list_id, owner_id, owner_screen_name, + endpoint_parameters=( + 'screen_name', 'user_id', 'slug', 'list_id', 'owner_id', + 'owner_screen_name' + ), **kwargs + ) + @payload('list') def remove_list_members(self, screen_name=None, user_id=None, slug=None, list_id=None, owner_id=None, - owner_screen_name=None): + owner_screen_name=None, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-members-destroy_all - :allowed_param: 'screen_name', 'user_id', 'slug', 'list_id', - 'owner_id', 'owner_screen_name' - """ - return bind_api( - api=self, - path='/lists/members/destroy_all.json', - method='POST', - payload_type='list', - allowed_param=['screen_name', 'user_id', 'slug', 'list_id', - 'owner_id', 'owner_screen_name'], - require_auth=True - )(list_to_csv(screen_name), list_to_csv(user_id), slug, list_id, - owner_id, owner_screen_name) - - @property - def list_members(self): + """ + return self.request( + 'POST', 'lists/members/destroy_all', list_to_csv(screen_name), + list_to_csv(user_id), slug, list_id, owner_id, owner_screen_name, + endpoint_parameters=( + 'screen_name', 'user_id', 'slug', 'list_id', 'owner_id', + 'owner_screen_name' + ), **kwargs + ) + + @pagination(mode='cursor') + @payload('user', list=True) + def list_members(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-members - :allowed_param: 'owner_screen_name', 'slug', 'list_id', 'owner_id', - 'cursor' """ - return bind_api( - api=self, - path='/lists/members.json', - payload_type='user', payload_list=True, - allowed_param=['owner_screen_name', 'slug', 'list_id', 'owner_id', - 'cursor'] + return self.request( + 'GET', 'lists/members', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'list_id', 'owner_id', 'cursor' + ), **kwargs ) - @property - def show_list_member(self): + @payload('user') + def show_list_member(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-members-show - :allowed_param: 'list_id', 'slug', 'user_id', 'screen_name', - 'owner_screen_name', 'owner_id' """ - return bind_api( - api=self, - path='/lists/members/show.json', - payload_type='user', - allowed_param=['list_id', 'slug', 'user_id', 'screen_name', - 'owner_screen_name', 'owner_id'] + return self.request( + 'GET', 'lists/members/show', *args, endpoint_parameters=( + 'list_id', 'slug', 'user_id', 'screen_name', + 'owner_screen_name', 'owner_id' + ), **kwargs ) - @property - def subscribe_list(self): + @payload('list') + def subscribe_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-subscribers-create - :allowed_param: 'owner_screen_name', 'slug', 'owner_id', 'list_id' """ - return bind_api( - api=self, - path='/lists/subscribers/create.json', - method='POST', - payload_type='list', - allowed_param=['owner_screen_name', 'slug', 'owner_id', 'list_id'], - require_auth=True + return self.request( + 'POST', 'lists/subscribers/create', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'owner_id', 'list_id' + ), **kwargs ) - @property - def unsubscribe_list(self): + @payload('list') + def unsubscribe_list(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-subscribers-destroy - :allowed_param: 'owner_screen_name', 'slug', 'owner_id', 'list_id' """ - return bind_api( - api=self, - path='/lists/subscribers/destroy.json', - method='POST', - payload_type='list', - allowed_param=['owner_screen_name', 'slug', 'owner_id', 'list_id'], - require_auth=True + return self.request( + 'POST', 'lists/subscribers/destroy', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'owner_id', 'list_id' + ), **kwargs ) - @property - def list_subscribers(self): + @pagination(mode='cursor') + @payload('user', list=True) + def list_subscribers(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-subscribers - :allowed_param: 'owner_screen_name', 'slug', 'owner_id', 'list_id', - 'cursor', 'count', 'include_entities', - 'skip_status' """ - return bind_api( - api=self, - path='/lists/subscribers.json', - payload_type='user', payload_list=True, - allowed_param=['owner_screen_name', 'slug', 'owner_id', 'list_id', - 'cursor', 'count', 'include_entities', - 'skip_status'] + return self.request( + 'GET', 'lists/subscribers', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'owner_id', 'list_id', 'cursor', + 'count', 'include_entities', 'skip_status' + ), **kwargs ) - @property - def show_list_subscriber(self): + @payload('user') + def show_list_subscriber(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/get-lists-subscribers-show - :allowed_param: 'owner_screen_name', 'slug', 'screen_name', - 'owner_id', 'list_id', 'user_id' """ - return bind_api( - api=self, - path='/lists/subscribers/show.json', - payload_type='user', - allowed_param=['owner_screen_name', 'slug', 'screen_name', - 'owner_id', 'list_id', 'user_id'] + return self.request( + 'GET', 'lists/subscribers/show', *args, endpoint_parameters=( + 'owner_screen_name', 'slug', 'screen_name', 'owner_id', + 'list_id', 'user_id' + ), **kwargs ) - @property - def trends_available(self): - """ :reference: https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-available """ - return bind_api( - api=self, - path='/trends/available.json', - payload_type='json' - ) + @payload('json') + def trends_available(self, *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-available + """ + return self.request('GET', 'trends/available', *args, **kwargs) - @property - def trends_place(self): + @payload('json') + def trends_place(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/trends/trends-for-location/api-reference/get-trends-place - :allowed_param: 'id', 'exclude' """ - return bind_api( - api=self, - path='/trends/place.json', - payload_type='json', - allowed_param=['id', 'exclude'] + return self.request( + 'GET', 'trends/place', *args, endpoint_parameters=( + 'id', 'exclude' + ), **kwargs ) - @property - def trends_closest(self): + @payload('json') + def trends_closest(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-closest - :allowed_param: 'lat', 'long' """ - return bind_api( - api=self, - path='/trends/closest.json', - payload_type='json', - allowed_param=['lat', 'long'] + return self.request( + 'GET', 'trends/closest', *args, endpoint_parameters=( + 'lat', 'long' + ), **kwargs ) - @property - def search(self): + @pagination(mode='id') + @payload('search_results') + def search(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/search/api-reference/get-search-tweets - :allowed_param: 'q', 'lang', 'locale', 'since_id', 'geocode', - 'max_id', 'until', 'result_type', 'count', - 'include_entities' """ - return bind_api( - api=self, - path='/search/tweets.json', - payload_type='search_results', - allowed_param=['q', 'lang', 'locale', 'since_id', 'geocode', - 'max_id', 'until', 'result_type', 'count', - 'include_entities'] + return self.request( + 'GET', 'search/tweets', *args, endpoint_parameters=( + 'q', 'lang', 'locale', 'since_id', 'geocode', 'max_id', + 'until', 'result_type', 'count', 'include_entities' + ), **kwargs ) @pagination(mode='next') + @payload('status', list=True) def search_30_day(self, environment_name, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/search/api-reference/premium-search - :allowed_param: 'query', 'tag', 'fromDate', 'toDate', 'maxResults', - 'next' - """ - return bind_api( - api=self, - path=f'/tweets/search/30day/{environment_name}.json', - payload_type='status', payload_list=True, - allowed_param=['query', 'tag', 'fromDate', 'toDate', 'maxResults', - 'next'], - require_auth=True - )(*args, **kwargs) + """ + return self.request( + 'GET', f'tweets/search/30day/{environment_name}', *args, + endpoint_parameters=( + 'query', 'tag', 'fromDate', 'toDate', 'maxResults', 'next' + ), **kwargs + ) @pagination(mode='next') + @payload('status', list=True) def search_full_archive(self, environment_name, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/tweets/search/api-reference/premium-search - :allowed_param: 'query', 'tag', 'fromDate', 'toDate', 'maxResults', - 'next' - """ - return bind_api( - api=self, - path=f'/tweets/search/fullarchive/{environment_name}.json', - payload_type='status', payload_list=True, - allowed_param=['query', 'tag', 'fromDate', 'toDate', 'maxResults', - 'next'], - require_auth=True - )(*args, **kwargs) - - @property - def reverse_geocode(self): + """ + return self.request( + 'GET', f'tweets/search/fullarchive/{environment_name}', *args, + endpoint_parameters=( + 'query', 'tag', 'fromDate', 'toDate', 'maxResults', 'next' + ), **kwargs + ) + + @payload('place', list=True) + def reverse_geocode(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/geo/places-near-location/api-reference/get-geo-reverse_geocode - :allowed_param: 'lat', 'long', 'accuracy', 'granularity', - 'max_results' """ - return bind_api( - api=self, - path='/geo/reverse_geocode.json', - payload_type='place', payload_list=True, - allowed_param=['lat', 'long', 'accuracy', 'granularity', - 'max_results'] + return self.request( + 'GET', 'geo/reverse_geocode', *args, endpoint_parameters=( + 'lat', 'long', 'accuracy', 'granularity', 'max_results' + ), **kwargs ) - @property - def geo_id(self): + @payload('place') + def geo_id(self, place_id, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/geo/place-information/api-reference/get-geo-id-place_id - :allowed_param: 'id' """ - return bind_api( - api=self, - path='/geo/id/{id}.json', - payload_type='place', - allowed_param=['id'] - ) + return self.request('GET', f'geo/id/{place_id}', *args, **kwargs) - @property - def geo_search(self): + @payload('place', list=True) + def geo_search(self, *args, **kwargs): """ :reference: https://developer.twitter.com/en/docs/geo/places-near-location/api-reference/get-geo-search - :allowed_param: 'lat', 'long', 'query', 'ip', 'granularity', - 'accuracy', 'max_results', 'contained_within' - """ - return bind_api( - api=self, - path='/geo/search.json', - payload_type='place', payload_list=True, - allowed_param=['lat', 'long', 'query', 'ip', 'granularity', - 'accuracy', 'max_results', 'contained_within'] + return self.request( + 'GET', 'geo/search', *args, endpoint_parameters=( + 'lat', 'long', 'query', 'ip', 'granularity', 'accuracy', + 'max_results', 'contained_within' + ), **kwargs ) - @property - def supported_languages(self): - """ :reference: https://developer.twitter.com/en/docs/developer-utilities/supported-languages/api-reference/get-help-languages """ - return bind_api( - api=self, - path='/help/languages.json', - payload_type='json', - require_auth=True - ) + @payload('json') + def supported_languages(self, *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/developer-utilities/supported-languages/api-reference/get-help-languages + """ + return self.request('GET', 'help/languages', *args, **kwargs) - @property - def configuration(self): - """ :reference: https://developer.twitter.com/en/docs/developer-utilities/configuration/api-reference/get-help-configuration """ - return bind_api( - api=self, - path='/help/configuration.json', - payload_type='json', - require_auth=True - ) + @payload('json') + def configuration(self, *args, **kwargs): + """ :reference: https://developer.twitter.com/en/docs/developer-utilities/configuration/api-reference/get-help-configuration + """ + return self.request('GET', 'help/configuration', *args, **kwargs) - - """ Internal use only """ - - @staticmethod - def _pack_image(filename, max_size, form_field='image', f=None, file_type=None): - """Pack image from file into multipart-formdata post body""" - # image must be less than 700kb in size - if f is None: - try: - if os.path.getsize(filename) > (max_size * 1024): - raise TweepError(f'File is too big, must be less than {max_size}kb.') - except os.error as e: - raise TweepError(f'Unable to access file: {e.strerror}') - - # build the mulitpart-formdata body - fp = open(filename, 'rb') - else: - f.seek(0, 2) # Seek to end of file - if f.tell() > (max_size * 1024): - raise TweepError(f'File is too big, must be less than {max_size}kb.') - f.seek(0) # Reset to beginning of file - fp = f - - # image must be gif, jpeg, png, webp - if not file_type: - h = None - if f is not None: - h = f.read(32) - f.seek(0) - file_type = imghdr.what(filename, h=h) or mimetypes.guess_type(filename)[0] - if file_type is None: - raise TweepError('Could not determine file type') - if file_type in ['gif', 'jpeg', 'png', 'webp']: - file_type = 'image/' + file_type - elif file_type not in ['image/gif', 'image/jpeg', 'image/png']: - raise TweepError(f'Invalid file type for image: {file_type}') - - if isinstance(filename, str): - filename = filename.encode('utf-8') - - BOUNDARY = b'Tw3ePy' - body = [] - body.append(b'--' + BOUNDARY) - body.append(f'Content-Disposition: form-data; name="{form_field}";' - f' filename="{filename}"' - .encode('utf-8')) - body.append(f'Content-Type: {file_type}'.encode('utf-8')) - body.append(b'') - body.append(fp.read()) - body.append(b'--' + BOUNDARY + b'--') - body.append(b'') - fp.close() - body = b'\r\n'.join(body) - - # build headers - headers = { - 'Content-Type': 'multipart/form-data; boundary=Tw3ePy', - 'Content-Length': str(len(body)) - } - - return headers, body