:mod:`tweepy.api` --- Twitter API wrapper
=========================================
-.. class:: API([auth_handler=None], [host='api.twitter.com'], [search_host='search.twitter.com'], [cache=None], [secure=False], [api_root='/1'], [search_root=''], [retry_count=0], [retry_delay=0], [retry_errors=None], [model_factory])
+.. class:: API([auth_handler=None], [host='api.twitter.com'], [search_host='search.twitter.com'], [cache=None], [api_root='/1'], [search_root=''], [retry_count=0], [retry_delay=0], [retry_errors=None], [model_factory])
This class provides a wrapper for the API as provided by
Twitter. The functions provided in this class are listed below.
:param host: general API host
:param search_host: search API host
:param cache: cache backend to use
- :param secure: if True use https
:param api_root: general API path root
:param search_root: search API path root
:param retry_count: default number of retries to attempt when error occurs
def __init__(self, auth_handler=None,
host='api.twitter.com', search_host='search.twitter.com',
- cache=None, secure=True, api_root='/1.1', search_root='',
+ cache=None, api_root='/1.1', search_root='',
retry_count=0, retry_delay=0, retry_errors=None, timeout=60,
parser=None, compression=False, wait_on_rate_limit=False,
wait_on_rate_limit_notify=False):
self.api_root = api_root
self.search_root = search_root
self.cache = cache
- self.secure = secure
self.compression = compression
self.retry_count = retry_count
self.retry_delay = retry_delay
OAUTH_HOST = 'api.twitter.com'
OAUTH_ROOT = '/oauth/'
- def __init__(self, consumer_key, consumer_secret, callback=None, secure=True):
+ def __init__(self, consumer_key, consumer_secret, callback=None):
if type(consumer_key) == unicode:
consumer_key = bytes(consumer_key)
self.access_token_secret = None
self.callback = callback
self.username = None
- self.secure = secure
self.oauth = OAuth1Session(consumer_key, client_secret=consumer_secret, callback_uri=self.callback)
- def _get_oauth_url(self, endpoint, secure=True):
- if self.secure or secure:
- prefix = 'https://'
- else:
- prefix = 'http://'
- return prefix + self.OAUTH_HOST + self.OAUTH_ROOT + endpoint
+ def _get_oauth_url(self, endpoint):
+ return 'https://' + self.OAUTH_HOST + self.OAUTH_ROOT + endpoint
def apply_auth(self):
return OAuth1(self.consumer_key, client_secret=self.consumer_secret, resource_owner_key=self.access_token, resource_owner_secret=self.access_token_secret)
and request activation of xAuth for it.
"""
try:
- url = self._get_oauth_url('access_token', secure=True)
+ url = self._get_oauth_url('access_token')
oauth = OAuth1(self.consumer_key, client_secret=self.consumer_secret)
r = requests.post(url=url, auth=oauth, headers={'x_auth_mode':
'client_auth', 'x_auth_username': username, 'x_auth_password':
OAUTH_HOST = 'api.twitter.com'
OAUTH_ROOT = '/oauth2/'
- def __init__(self, consumer_key, consumer_secret, callback=None, secure=True):
+ def __init__(self, consumer_key, consumer_secret, callback=None):
self.callback = callback
- self.secure = secure
self._bearer_token = ''
resp = requests.post(self.url, auth=(self.consumer_key, self.consumer_secret),
self._bearer_token = json_response['access_token']
- def _get_oauth_url(self, endpoint, secure=True):
- if self.secure or secure:
- prefix = 'https://'
- else:
- prefix = 'http://'
-
- return prefix + self.OAUTH_HOST + self.OAUTH_ROOT + endpoint
+ def _get_oauth_url(self, endpoint):
+ return 'https://' + self.OAUTH_HOST + self.OAUTH_ROOT + endpoint
def apply_auth(self):
# Perform any path variable substitution
self.build_path()
- if api.secure:
- self.scheme = 'https://'
- else:
- self.scheme = 'http://'
-
if self.search_api:
self.host = api.search_host
else:
# Build the request URL
url = self.api_root + self.path
- full_url = self.scheme + self.host + url
+ full_url = 'https://' + self.host + url
# Query the cache if one is available
# and this request uses a GET method.
self.snooze_time_step = options.get("snooze_time", 0.25)
self.snooze_time_cap = options.get("snooze_time_cap", 16)
self.buffer_size = options.get("buffer_size", 1500)
- if options.get("secure", True):
- self.scheme = "https"
- else:
- self.scheme = "http"
self.api = API()
self.session = requests.Session()
def _run(self):
# Authenticate
- url = "%s://%s%s" % (self.scheme, self.host, self.url)
+ url = "https://%s%s" % (self.host, self.url)
# Connect and process the stream
error_counter = 0