'exclude_replies', 'include_entities'
"""
return bind_api(
- api=self,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def statuses_lookup(self, id_, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-lookup
kwargs['map'] = kwargs.pop('map_')
return bind_api(
- api=self,
+ list_to_csv(id_), *args, 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)
+ require_auth=True, **kwargs
+ )
@pagination(mode='id')
def user_timeline(self, *args, **kwargs):
'include_rts'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'include_rts'], **kwargs
+ )
@pagination(mode='id')
def mentions_timeline(self, *args, **kwargs):
:allowed_param: 'since_id', 'max_id', 'count'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/statuses/mentions_timeline.json',
payload_type='status', payload_list=True,
allowed_param=['since_id', 'max_id', 'count'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def related_results(self, *args, **kwargs):
""" :reference: https://dev.twitter.com/docs/api/1.1/get/related_results/show/%3id.format
:allowed_param: 'id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/related_results/show/{id}.json',
payload_type='relation', payload_list=True,
allowed_param=['id'],
- require_auth=False
- )(*args, **kwargs)
+ require_auth=False, **kwargs
+ )
@pagination(mode='id')
def retweets_of_me(self, *args, **kwargs):
:allowed_param: 'since_id', 'max_id', 'count'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/statuses/retweets_of_me.json',
payload_type='status', payload_list=True,
allowed_param=['since_id', 'max_id', 'count'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def get_status(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-show-id
'include_card_uri'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'include_card_uri'], **kwargs
+ )
def update_status(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update
kwargs['media_ids'] = list_to_csv(kwargs['media_ids'])
return bind_api(
- api=self,
+ *args, api=self,
path='/statuses/update.json',
method='POST',
payload_type='status',
'place_id', 'display_coordinates', 'trim_user',
'enable_dmcommands', 'fail_dmcommands',
'card_uri'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def media_upload(self, filename, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/media/upload-media/api-reference/post-media-upload
kwargs.update({'headers': headers, 'post_data': post_data})
return bind_api(
- api=self,
+ *args, api=self,
path='/media/upload.json',
method='POST',
payload_type='media',
allowed_param=[],
require_auth=True,
- upload_api=True
- )(*args, **kwargs)
+ 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
}
return bind_api(
- api=self,
+ *args, api=self,
path='/media/metadata/create.json',
method='POST',
allowed_param=[],
require_auth=True,
- upload_api=True
- )(*args, **kwargs)
+ upload_api=True, **kwargs
+ )
def update_with_media(self, filename, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update_with_media
kwargs.update({'headers': headers, 'post_data': post_data})
return bind_api(
- api=self,
+ *args, api=self,
path='/statuses/update_with_media.json',
method='POST',
payload_type='status',
'in_reply_to_status_id_str',
'auto_populate_reply_metadata', 'lat', 'long',
'place_id', 'display_coordinates'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def destroy_status(self, *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,
+ *args, api=self,
path='/statuses/destroy/{id}.json',
method='POST',
payload_type='status',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def retweet(self, *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,
+ *args, api=self,
path='/statuses/retweet/{id}.json',
method='POST',
payload_type='status',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def unretweet(self, *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,
+ *args, api=self,
path='/statuses/unretweet/{id}.json',
method='POST',
payload_type='status',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def retweets(self, *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,
+ *args, api=self,
path='/statuses/retweets/{id}.json',
payload_type='status', payload_list=True,
allowed_param=['id', 'count'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def retweeters(self, *args, **kwargs):
:allowed_param: 'id', 'cursor', 'stringify_ids
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/statuses/retweeters/ids.json',
payload_type='ids',
- allowed_param=['id', 'cursor', 'stringify_ids']
- )(*args, **kwargs)
+ allowed_param=['id', 'cursor', 'stringify_ids'], **kwargs
+ )
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,
+ *args, api=self,
path='/users/show.json',
payload_type='user',
- allowed_param=['id', 'user_id', 'screen_name']
- )(*args, **kwargs)
+ allowed_param=['id', 'user_id', 'screen_name'], **kwargs
+ )
def get_oembed(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/get-statuses-oembed
'link_color', 'widget_type', 'dnt'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'link_color', 'widget_type', 'dnt'], **kwargs
+ )
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
'tweet_mode'
"""
return bind_api(
- api=self,
+ list_to_csv(user_ids), list_to_csv(screen_names), *args, 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)
+ 'tweet_mode'], **kwargs
+ )
def me(self):
""" Get the authenticated user """
:allowed_param: 'q', 'count', 'page'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/users/search.json',
payload_type='user', payload_list=True,
require_auth=True,
- allowed_param=['q', 'count', 'page']
- )(*args, **kwargs)
+ allowed_param=['q', 'count', 'page'], **kwargs
+ )
def get_direct_message(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/get-event
:allowed_param: 'id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/direct_messages/events/show.json',
payload_type='direct_message',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='dm_cursor')
def list_direct_messages(self, *args, **kwargs):
:allowed_param: 'count', 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/direct_messages/events/list.json',
payload_type='direct_message', payload_list=True,
allowed_param=['count', 'cursor'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def send_direct_message(self, recipient_id, text, quick_reply_options=None,
attachment_type=None, attachment_media_id=None,
path='/direct_messages/events/new.json',
method='POST',
payload_type='direct_message',
- require_auth=True
- )(json_payload=json_payload)
+ require_auth=True,
+ json_payload=json_payload
+ )
def destroy_direct_message(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/delete-message-event
:allowed_param: 'id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/direct_messages/events/destroy.json',
method='DELETE',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def create_friendship(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/post-friendships-create
:allowed_param: 'id', 'user_id', 'screen_name', 'follow'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friendships/create.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name', 'follow'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def destroy_friendship(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/post-friendships-destroy
:allowed_param: 'id', 'user_id', 'screen_name'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friendships/destroy.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def show_friendship(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-friendships-show
'target_screen_name'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friendships/show.json',
payload_type='friendship',
allowed_param=['source_id', 'source_screen_name',
- 'target_id', 'target_screen_name']
- )(*args, **kwargs)
+ 'target_id', 'target_screen_name'], **kwargs
+ )
def lookup_friendships(self, user_ids=None, screen_names=None):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-friendships-lookup
:allowed_param: 'user_id', 'screen_name'
"""
return bind_api(
- api=self,
+ list_to_csv(user_ids), list_to_csv(screen_names), api=self,
path='/friendships/lookup.json',
payload_type='relationship', payload_list=True,
allowed_param=['user_id', 'screen_name'],
require_auth=True
- )(list_to_csv(user_ids), list_to_csv(screen_names))
+ )
@pagination(mode='cursor')
def friends_ids(self, *args, **kwargs):
:allowed_param: 'id', 'user_id', 'screen_name', 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friends/ids.json',
payload_type='ids',
- allowed_param=['id', 'user_id', 'screen_name', 'cursor']
- )(*args, **kwargs)
+ allowed_param=['id', 'user_id', 'screen_name', 'cursor'], **kwargs
+ )
@pagination(mode='cursor')
def friends(self, *args, **kwargs):
'skip_status', 'include_user_entities'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friends/list.json',
payload_type='user', payload_list=True,
allowed_param=['id', 'user_id', 'screen_name', 'cursor', 'count',
- 'skip_status', 'include_user_entities']
- )(*args, **kwargs)
+ 'skip_status', 'include_user_entities'], **kwargs
+ )
@pagination(mode='cursor')
def friendships_incoming(self, *args, **kwargs):
:allowed_param: 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friendships/incoming.json',
payload_type='ids',
- allowed_param=['cursor']
- )(*args, **kwargs)
+ allowed_param=['cursor'], **kwargs
+ )
@pagination(mode='cursor')
def friendships_outgoing(self, *args, **kwargs):
:allowed_param: 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/friendships/outgoing.json',
payload_type='ids',
- allowed_param=['cursor']
- )(*args, **kwargs)
+ allowed_param=['cursor'], **kwargs
+ )
@pagination(mode='cursor')
def followers_ids(self, *args, **kwargs):
:allowed_param: 'id', 'user_id', 'screen_name', 'cursor', 'count'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/followers/ids.json',
payload_type='ids',
- allowed_param=['id', 'user_id', 'screen_name', 'cursor', 'count']
- )(*args, **kwargs)
+ allowed_param=['id', 'user_id', 'screen_name', 'cursor', 'count'],
+ **kwargs
+ )
@pagination(mode='cursor')
def followers(self, *args, **kwargs):
'skip_status', 'include_user_entities'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/followers/list.json',
payload_type='user', payload_list=True,
allowed_param=['id', 'user_id', 'screen_name', 'cursor', 'count',
- 'skip_status', 'include_user_entities']
- )(*args, **kwargs)
+ 'skip_status', 'include_user_entities'], **kwargs
+ )
def get_settings(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/get-account-settings """
return bind_api(
- api=self,
+ *args, api=self,
path='/account/settings.json',
payload_type='json',
- use_cache=False
- )(*args, **kwargs)
+ use_cache=False, **kwargs
+ )
def set_settings(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-settings
'allow_contributor_request', 'lang'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/account/settings.json',
method='POST',
payload_type='json',
'end_sleep_time', 'time_zone',
'trend_location_woeid',
'allow_contributor_request', 'lang'],
- use_cache=False
- )(*args, **kwargs)
+ use_cache=False, **kwargs
+ )
def verify_credentials(self, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/get-account-verify_credentials
payload_type='user',
require_auth=True,
allowed_param=['include_entities', 'skip_status',
- 'include_email'],
- )(**kwargs)
+ 'include_email'], **kwargs
+ )
except TweepError as e:
if e.response is not None and e.response.status_code == 401:
return False
:allowed_param: 'resources'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/application/rate_limit_status.json',
payload_type='json',
allowed_param=['resources'],
- use_cache=False
- )(*args, **kwargs)
+ use_cache=False, **kwargs
+ )
def update_profile_image(self, filename, file_=None):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-update_profile_image
"""
headers, post_data = API._pack_image(filename, 700, f=file_)
return bind_api(
- api=self,
+ self, api=self,
path='/account/update_profile_image.json',
method='POST',
payload_type='user',
allowed_param=['include_entities', 'skip_status'],
- require_auth=True
- )(self, post_data=post_data, headers=headers)
+ require_auth=True,
+ post_data=post_data, headers=headers
+ )
def update_profile_banner(self, filename, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/manage-account-settings/api-reference/post-account-update_profile_banner
path='/account/update_profile_banner.json',
method='POST',
allowed_param=['width', 'height', 'offset_left', 'offset_right'],
- require_auth=True
- )(post_data=post_data, headers=headers)
+ require_auth=True,
+ post_data=post_data, headers=headers
+ )
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
'profile_link_color'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/account/update_profile.json',
method='POST',
payload_type='user',
allowed_param=['name', 'url', 'location', 'description',
'profile_link_color'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='id')
def favorites(self, *args, **kwargs):
'since_id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/favorites/list.json',
payload_type='status', payload_list=True,
allowed_param=['screen_name', 'user_id', 'max_id', 'count',
- 'since_id']
- )(*args, **kwargs)
+ 'since_id'], **kwargs
+ )
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,
+ *args, api=self,
path='/favorites/create.json',
method='POST',
payload_type='status',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/favorites/destroy.json',
method='POST',
payload_type='status',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/blocks/create.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/blocks/destroy.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def mutes_ids(self, *args, **kwargs):
:allowed_param: 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/mutes/users/ids.json',
payload_type='ids',
allowed_param=['cursor'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def mutes(self, *args, **kwargs):
:allowed_param: 'cursor', 'include_entities', 'skip_status'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/mutes/users/list.json',
payload_type='user', payload_list=True,
allowed_param=['cursor', 'include_entities', 'skip_status'],
- required_auth=True
- )(*args, **kwargs)
+ required_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/mutes/users/create.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/mutes/users/destroy.json',
method='POST',
payload_type='user',
allowed_param=['id', 'user_id', 'screen_name'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def blocks(self, *args, **kwargs):
:allowed_param: 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/blocks/list.json',
payload_type='user', payload_list=True,
allowed_param=['cursor'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def blocks_ids(self, *args, **kwargs):
:allowed_param: 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/blocks/ids.json',
payload_type='ids',
allowed_param=['cursor'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/users/report_spam.json',
method='POST',
payload_type='user',
allowed_param=['user_id', 'screen_name', 'perform_block'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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 bind_api(
- api=self,
+ *args, api=self,
path='/saved_searches/list.json',
payload_type='saved_search', payload_list=True,
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def get_saved_search(self, *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,
+ *args, api=self,
path='/saved_searches/show/{id}.json',
payload_type='saved_search',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/saved_searches/create.json',
method='POST',
payload_type='saved_search',
allowed_param=['query'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def destroy_saved_search(self, *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,
+ *args, api=self,
path='/saved_searches/destroy/{id}.json',
method='POST',
payload_type='saved_search',
allowed_param=['id'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/lists/create.json',
method='POST',
payload_type='list',
allowed_param=['name', 'mode', 'description'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/lists/destroy.json',
method='POST',
payload_type='list',
allowed_param=['owner_screen_name', 'owner_id', 'list_id', 'slug'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def update_list(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/accounts-and-users/create-manage-lists/api-reference/post-lists-update
'owner_screen_name', 'owner_id'
"""
return bind_api(
- api=self,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, api=self,
path='/lists/list.json',
payload_type='list', payload_list=True,
allowed_param=['screen_name', 'user_id', 'reverse'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def lists_memberships(self, *args, **kwargs):
'cursor', 'count'
"""
return bind_api(
- api=self,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def lists_ownerships(self, *args, **kwargs):
:allowed_param: 'user_id', 'screen_name', 'count', 'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/lists/ownerships.json',
payload_type='list', payload_list=True,
allowed_param=['user_id', 'screen_name', 'count', 'cursor'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def lists_subscriptions(self, *args, **kwargs):
:allowed_param: 'screen_name', 'user_id', 'cursor', 'count'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/lists/subscriptions.json',
payload_type='list', payload_list=True,
allowed_param=['screen_name', 'user_id', 'cursor', 'count'],
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='id')
def list_timeline(self, *args, **kwargs):
'include_rts'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'include_rts'], **kwargs
+ )
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,
+ *args, api=self,
path='/lists/show.json',
payload_type='list',
- allowed_param=['owner_screen_name', 'owner_id', 'slug', 'list_id']
- )(*args, **kwargs)
+ allowed_param=['owner_screen_name', 'owner_id', 'slug', 'list_id'],
+ **kwargs
+ )
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
'owner_id', 'slug', 'list_id'
"""
return bind_api(
- api=self,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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
'owner_id', 'slug', 'list_id'
"""
return bind_api(
- api=self,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def add_list_members(self, screen_name=None, user_id=None, slug=None,
list_id=None, owner_id=None, owner_screen_name=None):
'owner_id', 'owner_screen_name'
"""
return bind_api(
+ list_to_csv(screen_name), list_to_csv(user_id), slug, list_id,
+ owner_id, owner_screen_name,
api=self,
path='/lists/members/create_all.json',
method='POST',
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)
+ )
def remove_list_members(self, screen_name=None, user_id=None, slug=None,
list_id=None, owner_id=None,
'owner_id', 'owner_screen_name'
"""
return bind_api(
+ list_to_csv(screen_name), list_to_csv(user_id), slug, list_id,
+ owner_id, owner_screen_name,
api=self,
path='/lists/members/destroy_all.json',
method='POST',
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)
+ )
@pagination(mode='cursor')
def list_members(self, *args, **kwargs):
'cursor'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/lists/members.json',
payload_type='user', payload_list=True,
allowed_param=['owner_screen_name', 'slug', 'list_id', 'owner_id',
- 'cursor']
- )(*args, **kwargs)
+ 'cursor'], **kwargs
+ )
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
'owner_screen_name', 'owner_id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/lists/members/show.json',
payload_type='user',
allowed_param=['list_id', 'slug', 'user_id', 'screen_name',
- 'owner_screen_name', 'owner_id']
- )(*args, **kwargs)
+ 'owner_screen_name', 'owner_id'], **kwargs
+ )
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,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
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,
+ *args, 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
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
@pagination(mode='cursor')
def list_subscribers(self, *args, **kwargs):
'skip_status'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'skip_status'], **kwargs
+ )
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
'owner_id', 'list_id', 'user_id'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/lists/subscribers/show.json',
payload_type='user',
allowed_param=['owner_screen_name', 'slug', 'screen_name',
- 'owner_id', 'list_id', 'user_id']
- )(*args, **kwargs)
+ 'owner_id', 'list_id', 'user_id'], **kwargs
+ )
def trends_available(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-available """
return bind_api(
- api=self,
+ *args, api=self,
path='/trends/available.json',
- payload_type='json'
- )(*args, **kwargs)
+ payload_type='json', **kwargs
+ )
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,
+ *args, api=self,
path='/trends/place.json',
payload_type='json',
- allowed_param=['id', 'exclude']
- )(*args, **kwargs)
+ allowed_param=['id', 'exclude'], **kwargs
+ )
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,
+ *args, api=self,
path='/trends/closest.json',
payload_type='json',
- allowed_param=['lat', 'long']
- )(*args, **kwargs)
+ allowed_param=['lat', 'long'], **kwargs
+ )
@pagination(mode='id')
def search(self, *args, **kwargs):
'include_entities'
"""
return bind_api(
- api=self,
+ *args, 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']
- )(*args, **kwargs)
+ 'include_entities'], **kwargs
+ )
@pagination(mode='next')
def search_30_day(self, environment_name, *args, **kwargs):
'next'
"""
return bind_api(
- api=self,
+ *args, 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)
+ require_auth=True, **kwargs
+ )
@pagination(mode='next')
def search_full_archive(self, environment_name, *args, **kwargs):
'next'
"""
return bind_api(
- api=self,
+ *args, 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)
+ require_auth=True, **kwargs
+ )
def reverse_geocode(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/geo/places-near-location/api-reference/get-geo-reverse_geocode
'max_results'
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/geo/reverse_geocode.json',
payload_type='place', payload_list=True,
allowed_param=['lat', 'long', 'accuracy', 'granularity',
- 'max_results']
- )(*args, **kwargs)
+ 'max_results'], **kwargs
+ )
def geo_id(self, *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,
+ *args, api=self,
path='/geo/id/{id}.json',
payload_type='place',
- allowed_param=['id']
- )(*args, **kwargs)
+ allowed_param=['id'], **kwargs
+ )
def geo_search(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/geo/places-near-location/api-reference/get-geo-search
"""
return bind_api(
- api=self,
+ *args, api=self,
path='/geo/search.json',
payload_type='place', payload_list=True,
allowed_param=['lat', 'long', 'query', 'ip', 'granularity',
- 'accuracy', 'max_results', 'contained_within']
- )(*args, **kwargs)
+ 'accuracy', 'max_results', 'contained_within'],
+ **kwargs
+ )
def supported_languages(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/developer-utilities/supported-languages/api-reference/get-help-languages """
return bind_api(
- api=self,
+ *args, api=self,
path='/help/languages.json',
payload_type='json',
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
def configuration(self, *args, **kwargs):
""" :reference: https://developer.twitter.com/en/docs/developer-utilities/configuration/api-reference/get-help-configuration """
return bind_api(
- api=self,
+ *args, api=self,
path='/help/configuration.json',
payload_type='json',
- require_auth=True
- )(*args, **kwargs)
+ require_auth=True, **kwargs
+ )
""" Internal use only """
log = logging.getLogger(__name__)
-def bind_api(**config):
-
- class APIMethod:
-
- api = config['api']
- path = config['path']
- payload_type = config.get('payload_type', None)
- payload_list = config.get('payload_list', False)
- allowed_param = config.get('allowed_param', [])
- method = config.get('method', 'GET')
- require_auth = config.get('require_auth', False)
- upload_api = config.get('upload_api', False)
- use_cache = config.get('use_cache', True)
- session = requests.Session()
-
- def __init__(self, *args, **kwargs):
- api = self.api
- # If authentication is required and no credentials
- # are provided, throw an error.
- if self.require_auth and not api.auth:
- raise TweepError('Authentication required!')
-
- self.post_data = kwargs.pop('post_data', None)
- self.json_payload = kwargs.pop('json_payload', None)
- self.retry_count = kwargs.pop('retry_count',
- api.retry_count)
- self.retry_delay = kwargs.pop('retry_delay',
- api.retry_delay)
- self.retry_errors = kwargs.pop('retry_errors',
- api.retry_errors)
- self.wait_on_rate_limit = kwargs.pop('wait_on_rate_limit',
- api.wait_on_rate_limit)
- self.wait_on_rate_limit_notify = kwargs.pop('wait_on_rate_limit_notify',
- api.wait_on_rate_limit_notify)
- self.return_cursors = kwargs.pop('return_cursors', False)
- self.parser = kwargs.pop('parser', api.parser)
- self.headers = kwargs.pop('headers', {})
- self.build_parameters(args, kwargs)
-
- # Pick correct URL root to use
- if self.upload_api:
- self.api_root = api.upload_root
+class APIMethod:
+
+ def __init__(self, *args, **kwargs):
+ self.api = api = kwargs.pop('api')
+ self.path = kwargs.pop('path')
+ self.payload_type = kwargs.pop('payload_type', None)
+ self.payload_list = kwargs.pop('payload_list', False)
+ self.allowed_param = kwargs.pop('allowed_param', [])
+ self.method = kwargs.pop('method', 'GET')
+ self.require_auth = kwargs.pop('require_auth', False)
+ self.upload_api = kwargs.pop('upload_api', False)
+ self.use_cache = kwargs.pop('use_cache', True)
+ self.session = requests.Session()
+
+ # If authentication is required and no credentials
+ # are provided, throw an error.
+ if self.require_auth and not api.auth:
+ raise TweepError('Authentication required!')
+
+ self.post_data = kwargs.pop('post_data', None)
+ self.json_payload = kwargs.pop('json_payload', None)
+ self.retry_count = kwargs.pop('retry_count',
+ api.retry_count)
+ self.retry_delay = kwargs.pop('retry_delay',
+ api.retry_delay)
+ self.retry_errors = kwargs.pop('retry_errors',
+ api.retry_errors)
+ self.wait_on_rate_limit = kwargs.pop('wait_on_rate_limit',
+ api.wait_on_rate_limit)
+ self.wait_on_rate_limit_notify = kwargs.pop('wait_on_rate_limit_notify',
+ api.wait_on_rate_limit_notify)
+ self.return_cursors = kwargs.pop('return_cursors', False)
+ self.parser = kwargs.pop('parser', api.parser)
+ self.headers = kwargs.pop('headers', {})
+ self.build_parameters(args, kwargs)
+
+ # Pick correct URL root to use
+ if self.upload_api:
+ self.api_root = api.upload_root
+ else:
+ self.api_root = api.api_root
+
+ # Perform any path variable substitution
+ self.build_path()
+
+ if self.upload_api:
+ self.host = api.upload_host
+ else:
+ self.host = api.host
+
+ # Monitoring rate limits
+ self._remaining_calls = None
+ self._reset_time = None
+
+ def build_parameters(self, args, kwargs):
+ self.session.params = {}
+ for idx, arg in enumerate(args):
+ if arg is None:
+ continue
+ try:
+ self.session.params[self.allowed_param[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 self.session.params:
+ raise TweepError(f'Multiple values for parameter {k} supplied!')
+
+ self.session.params[k] = str(arg)
+
+ log.debug("PARAMS: %r", self.session.params)
+
+ def build_path(self):
+ for variable in re_path_template.findall(self.path):
+ name = variable.strip('{}')
+
+ if name == 'user' and 'user' not in self.session.params and self.api.auth:
+ # No 'user' parameter provided, fetch it from Auth instead.
+ value = self.api.auth.get_username()
else:
- self.api_root = api.api_root
-
- # Perform any path variable substitution
- self.build_path()
-
- if self.upload_api:
- self.host = api.upload_host
- else:
- self.host = api.host
-
- # Monitoring rate limits
- self._remaining_calls = None
- self._reset_time = None
-
- def build_parameters(self, args, kwargs):
- self.session.params = {}
- for idx, arg in enumerate(args):
- if arg is None:
- continue
- try:
- self.session.params[self.allowed_param[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 self.session.params:
- raise TweepError(f'Multiple values for parameter {k} supplied!')
-
- self.session.params[k] = str(arg)
-
- log.debug("PARAMS: %r", self.session.params)
-
- def build_path(self):
- for variable in re_path_template.findall(self.path):
- name = variable.strip('{}')
-
- if name == 'user' and 'user' not in self.session.params and self.api.auth:
- # No 'user' parameter provided, fetch it from Auth instead.
- value = self.api.auth.get_username()
- else:
- try:
- value = quote(self.session.params[name])
- except KeyError:
- raise TweepError(f'No parameter value found for path variable: {name}')
- del self.session.params[name]
-
- self.path = self.path.replace(variable, value)
-
- def execute(self):
- self.api.cached_result = False
-
- # Build the request URL
- url = self.api_root + self.path
- full_url = 'https://' + self.host + url
-
- # Query the cache if one is available
- # and this request uses a GET method.
- if self.use_cache and self.api.cache and self.method == 'GET':
- cache_result = self.api.cache.get(f'{url}?{urlencode(self.session.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.api
- else:
- if isinstance(cache_result, Model):
- cache_result._api = self.api
- self.api.cached_result = True
- return cache_result
-
- # Continue attempting request until successful
- # or maximum number of retries is reached.
- retries_performed = 0
- while retries_performed < self.retry_count + 1:
- if (self.wait_on_rate_limit and self._reset_time is not None
- and self._remaining_calls is not None
- and self._remaining_calls < 1):
- # Handle running out of API calls
- sleep_time = self._reset_time - int(time.time())
- if sleep_time > 0:
- if self.wait_on_rate_limit_notify:
- 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.api.auth:
- auth = self.api.auth.apply_auth()
-
- # Execute request
try:
- resp = self.session.request(self.method,
- full_url,
- headers=self.headers,
- data=self.post_data,
- json=self.json_payload,
- timeout=self.api.timeout,
- auth=auth,
- proxies=self.api.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:
- self._remaining_calls = int(rem_calls)
- elif self._remaining_calls is not None:
- self._remaining_calls -= 1
-
- reset_time = resp.headers.get('x-rate-limit-reset')
- if reset_time is not None:
- self._reset_time = int(reset_time)
-
- retry_delay = self.retry_delay
- if resp.status_code in (420, 429) and self.wait_on_rate_limit:
- if self._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.api.last_response = resp
- if resp.status_code and not 200 <= resp.status_code < 300:
- try:
- error_msg, api_error_code = \
- self.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)
+ value = quote(self.session.params[name])
+ except KeyError:
+ raise TweepError(f'No parameter value found for path variable: {name}')
+ del self.session.params[name]
+
+ self.path = self.path.replace(variable, value)
+
+ def execute(self):
+ self.api.cached_result = False
+
+ # Build the request URL
+ url = self.api_root + self.path
+ full_url = 'https://' + self.host + url
+
+ # Query the cache if one is available
+ # and this request uses a GET method.
+ if self.use_cache and self.api.cache and self.method == 'GET':
+ cache_result = self.api.cache.get(f'{url}?{urlencode(self.session.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.api
else:
- raise TweepError(error_msg, resp, api_code=api_error_code)
+ if isinstance(cache_result, Model):
+ cache_result._api = self.api
+ self.api.cached_result = True
+ return cache_result
+
+ # Continue attempting request until successful
+ # or maximum number of retries is reached.
+ retries_performed = 0
+ while retries_performed < self.retry_count + 1:
+ if (self.wait_on_rate_limit and self._reset_time is not None
+ and self._remaining_calls is not None
+ and self._remaining_calls < 1):
+ # Handle running out of API calls
+ sleep_time = self._reset_time - int(time.time())
+ if sleep_time > 0:
+ if self.wait_on_rate_limit_notify:
+ 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.api.auth:
+ auth = self.api.auth.apply_auth()
+
+ # Execute request
+ try:
+ resp = self.session.request(self.method,
+ full_url,
+ headers=self.headers,
+ data=self.post_data,
+ json=self.json_payload,
+ timeout=self.api.timeout,
+ auth=auth,
+ proxies=self.api.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:
+ self._remaining_calls = int(rem_calls)
+ elif self._remaining_calls is not None:
+ self._remaining_calls -= 1
+
+ reset_time = resp.headers.get('x-rate-limit-reset')
+ if reset_time is not None:
+ self._reset_time = int(reset_time)
+
+ retry_delay = self.retry_delay
+ if resp.status_code in (420, 429) and self.wait_on_rate_limit:
+ if self._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.api.last_response = resp
+ if resp.status_code and not 200 <= resp.status_code < 300:
+ try:
+ error_msg, api_error_code = \
+ self.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
- self.return_cursors = (self.return_cursors or
- 'cursor' in self.session.params or 'next' in self.session.params)
- result = self.parser.parse(self, resp.text, return_cursors=self.return_cursors)
+ # Parse the response payload
+ self.return_cursors = (self.return_cursors or
+ 'cursor' in self.session.params or 'next' in self.session.params)
+ result = self.parser.parse(self, resp.text, return_cursors=self.return_cursors)
- # Store result into cache if one is available.
- if self.use_cache and self.api.cache and self.method == 'GET' and result:
- self.api.cache.store(f'{url}?{urlencode(self.session.params)}', result)
+ # Store result into cache if one is available.
+ if self.use_cache and self.api.cache and self.method == 'GET' and result:
+ self.api.cache.store(f'{url}?{urlencode(self.session.params)}', result)
- return result
+ return result
- def _call(*args, **kwargs):
- method = APIMethod(*args, **kwargs)
- try:
- if kwargs.get('create'):
- return method
- else:
- return method.execute()
- finally:
- method.session.close()
- return _call
+def bind_api(*args, **kwargs):
+ method = APIMethod(*args, **kwargs)
+ try:
+ if kwargs.get('create'):
+ return method
+ else:
+ return method.execute()
+ finally:
+ method.session.close()
def pagination(mode):
projects / tweepy.git / commitdiff