import requests
from tweepy.error import is_rate_limit_error_message, RateLimitError, TweepError
+from tweepy.errors import TweepyException
from tweepy.models import Model
from tweepy.parsers import ModelParser, Parser
from tweepy.utils import list_to_csv
# If authentication is required and no credentials
# are provided, throw an error.
if require_auth and not self.auth:
- raise TweepError('Authentication required!')
+ raise TweepyException('Authentication required!')
self.cached_result = False
auth=auth, proxies=self.proxy
)
except Exception as e:
- raise TweepError(f'Failed to send request: {e}').with_traceback(sys.exc_info()[2])
+ raise TweepyException(f'Failed to send request: {e}').with_traceback(sys.exc_info()[2])
if 200 <= resp.status_code < 300:
break
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.')
+ raise TweepyException(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}')
+ raise TweepyException(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.')
+ raise TweepyException(f'File is too big, must be less than {max_size}kb.')
f.seek(0) # Reset to beginning of file
fp = f
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')
+ raise TweepyException('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}')
+ raise TweepyException(f'Invalid file type for image: {file_type}')
if isinstance(filename, str):
filename = filename.encode('utf-8')
from requests_oauthlib import OAuth1, OAuth1Session
from tweepy.api import API
-from tweepy.error import TweepError
+from tweepy.errors import TweepyException
WARNING_MESSAGE = """Warning! Due to a Twitter API bug, signin_with_twitter
and access_type don't always play nice together. Details
url += f'?x_auth_access_type={access_type}'
return self.oauth.fetch_request_token(url)
except Exception as e:
- raise TweepError(e)
+ raise TweepyException(e)
def set_access_token(self, key, secret):
self.access_token = key
self.request_token = self._get_request_token(access_type=access_type)
return self.oauth.authorization_url(url)
except Exception as e:
- raise TweepError(e)
+ raise TweepyException(e)
def get_access_token(self, verifier=None):
"""
self.access_token_secret = resp['oauth_token_secret']
return self.access_token, self.access_token_secret
except Exception as e:
- raise TweepError(e)
+ raise TweepyException(e)
def get_xauth_access_token(self, username, password):
"""
credentials = parse_qs(r.content)
return credentials.get('oauth_token')[0], credentials.get('oauth_token_secret')[0]
except Exception as e:
- raise TweepError(e)
+ raise TweepyException(e)
def get_username(self):
if self.username is None:
if user:
self.username = user.screen_name
else:
- raise TweepError('Unable to get username,'
- ' invalid oauth token!')
+ raise TweepyException('Unable to get username,'
+ ' invalid oauth token!')
return self.username
data={'grant_type': 'client_credentials'})
data = resp.json()
if data.get('token_type') != 'bearer':
- raise TweepError('Expected token_type to equal "bearer", '
- f'but got {data.get("token_type")} instead')
+ raise TweepyException('Expected token_type to equal "bearer", '
+ f'but got {data.get("token_type")} instead')
self._bearer_token = data['access_token']
# Copyright 2009-2021 Joshua Roesslein
# See LICENSE for details.
-from tweepy.error import TweepError
+from tweepy.errors import TweepyException
from tweepy.parsers import ModelParser, RawParser
elif method.pagination_mode == 'page':
self.iterator = PageIterator(method, *args, **kwargs)
else:
- raise TweepError('Invalid pagination mode.')
+ raise TweepyException('Invalid pagination mode.')
else:
- raise TweepError('This method does not perform pagination')
+ raise TweepyException('This method does not perform pagination')
def pages(self, limit=0):
"""Return iterator for pages"""
def prev(self):
if self.prev_cursor == 0:
- raise TweepError('Can not page back more, at first page')
+ raise TweepyException('Can not page back more, at first page')
data, self.next_cursor, self.prev_cursor = self.method(cursor=self.prev_cursor,
*self.args,
**self.kwargs)
return data
def prev(self):
- raise TweepError('This method does not allow backwards pagination')
+ raise TweepyException('This method does not allow backwards pagination')
class IdIterator(BaseIterator):
def prev(self):
if self.current_page == 1:
- raise TweepError('Can not page back more, at first page')
+ raise TweepyException('Can not page back more, at first page')
self.current_page -= 1
return self.method(page=self.current_page, *self.args, **self.kwargs)
return data
def prev(self):
- raise TweepError('This method does not allow backwards pagination')
+ raise TweepyException('This method does not allow backwards pagination')
class ItemIterator(BaseIterator):
def prev(self):
if self.current_page is None:
- raise TweepError('Can not go back more, at first page')
+ raise TweepyException('Can not go back more, at first page')
if self.page_index == 0:
# At the beginning of the current page, move to next...
self.current_page = self.page_iterator.prev()
self.page_index = len(self.current_page)
if self.page_index == 0:
- raise TweepError('No more items')
+ raise TweepyException('No more items')
self.page_index -= 1
self.num_tweets -= 1
return self.current_page[self.page_index]
import json as json_lib
-from tweepy.error import TweepError
+from tweepy.errors import TweepyException
from tweepy.models import ModelFactory
try:
json = json_lib.loads(payload)
except Exception as e:
- raise TweepError(f'Failed to parse JSON payload: {e}')
+ raise TweepyException(f'Failed to parse JSON payload: {e}')
if return_cursors and isinstance(json, dict):
if 'next' in json:
return
model = getattr(self.model_factory, payload_type)
except AttributeError:
- raise TweepError(f'No model for this payload type: {payload_type}')
+ raise TweepyException(
+ f'No model for this payload type: {payload_type}'
+ )
json = JSONParser.parse(self, payload, return_cursors=return_cursors)
if isinstance(json, tuple):
import urllib3
import tweepy
-from tweepy.error import TweepError
+from tweepy.errors import TweepyException
from tweepy.models import Status
log = logging.getLogger(__name__)
filter_level=None, languages=None, stall_warnings=False,
threaded=False):
if self.running:
- raise TweepError("Stream is already connected")
+ raise TweepyException("Stream is already connected")
method = "POST"
endpoint = "statuses/filter"
body["track"] = ','.join(map(str, track))
if locations and len(locations) > 0:
if len(locations) % 4:
- raise TweepError(
+ raise TweepyException(
"Number of location coordinates should be a multiple of 4"
)
body["locations"] = ','.join(f"{l:.4f}" for l in locations)
def sample(self, *, languages=None, stall_warnings=False, threaded=False):
if self.running:
- raise TweepError("Stream is already connected")
+ raise TweepyException("Stream is already connected")
method = "GET"
endpoint = "statuses/sample"
projects / tweepy.git / commitdiff