#!/usr/bin/env python
#
#
# Copyright 2007-2016 The Python-Twitter Developers
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""A library that provides a Python interface to the Twitter API"""
from __future__ import division
from __future__ import print_function
import json
import sys
import gzip
import time
import base64
import re
import requests
from requests_oauthlib import OAuth1, OAuth2
import io
import warnings
from uuid import uuid4
import os
try:
# python 3
from urllib.parse import urlparse, urlunparse, urlencode
from urllib.request import urlopen
from urllib.request import __version__ as urllib_version
except ImportError:
from urlparse import urlparse, urlunparse
from urllib2 import urlopen
from urllib import urlencode
from urllib import __version__ as urllib_version
from twitter import (
__version__,
_FileCache,
Category,
DirectMessage,
List,
Status,
Trend,
User,
UserStatus,
)
from twitter.ratelimit import RateLimit
from twitter.twitter_utils import (
calc_expected_status_length,
is_url,
parse_media_file,
enf_type)
from twitter.error import (
TwitterError,
PythonTwitterDeprecationWarning330,
)
if sys.version_info > (3,):
long = int
CHARACTER_LIMIT = 140
# A singleton representing a lazily instantiated FileCache.
DEFAULT_CACHE = object()
[docs]class Api(object):
"""A python interface into the Twitter API
By default, the Api caches results for 1 minute.
Example usage:
To create an instance of the twitter.Api class, with no authentication:
>>> import twitter
>>> api = twitter.Api()
To fetch a single user's public status messages, where "user" is either
a Twitter "short name" or their user id.
>>> statuses = api.GetUserTimeline(user)
>>> print([s.text for s in statuses])
To use authentication, instantiate the twitter.Api class with a
consumer key and secret; and the oAuth key and secret:
>>> api = twitter.Api(consumer_key='twitter consumer key',
consumer_secret='twitter consumer secret',
access_token_key='the_key_given',
access_token_secret='the_key_secret')
To fetch your friends (after being authenticated):
>>> users = api.GetFriends()
>>> print([u.name for u in users])
To post a twitter status message (after being authenticated):
>>> status = api.PostUpdate('I love python-twitter!')
>>> print(status.text)
I love python-twitter!
There are many other methods, including:
>>> api.PostUpdates(status)
>>> api.PostDirectMessage(user, text)
>>> api.GetUser(user)
>>> api.GetReplies()
>>> api.GetUserTimeline(user)
>>> api.GetHomeTimeline()
>>> api.GetStatus(status_id)
>>> api.DestroyStatus(status_id)
>>> api.GetFriends(user)
>>> api.GetFollowers()
>>> api.GetFeatured()
>>> api.GetDirectMessages()
>>> api.GetSentDirectMessages()
>>> api.PostDirectMessage(user, text)
>>> api.DestroyDirectMessage(message_id)
>>> api.DestroyFriendship(user)
>>> api.CreateFriendship(user)
>>> api.LookupFriendship(user)
>>> api.VerifyCredentials()
"""
DEFAULT_CACHE_TIMEOUT = 60 # cache for 1 minute
_API_REALM = 'Twitter API'
def __init__(self,
consumer_key=None,
consumer_secret=None,
access_token_key=None,
access_token_secret=None,
application_only_auth=False,
input_encoding=None,
request_headers=None,
cache=DEFAULT_CACHE,
base_url=None,
stream_url=None,
upload_url=None,
chunk_size=1024 * 1024,
use_gzip_compression=False,
debugHTTP=False,
timeout=None,
sleep_on_rate_limit=False,
tweet_mode='compat',
proxies=None):
"""Instantiate a new twitter.Api object.
Args:
consumer_key (str):
Your Twitter user's consumer_key.
consumer_secret (str):
Your Twitter user's consumer_secret.
access_token_key (str):
The oAuth access token key value you retrieved
from running get_access_token.py.
access_token_secret (str):
The oAuth access token's secret, also retrieved
from the get_access_token.py run.
application_only_auth:
Use Application-Only Auth instead of User Auth.
Defaults to False [Optional]
input_encoding (str, optional):
The encoding used to encode input strings.
request_header (dict, optional):
A dictionary of additional HTTP request headers.
cache (object, optional):
The cache instance to use. Defaults to DEFAULT_CACHE.
Use None to disable caching.
base_url (str, optional):
The base URL to use to contact the Twitter API.
Defaults to https://api.twitter.com.
stream_url (str, optional):
The base URL to use for streaming endpoints.
Defaults to 'https://stream.twitter.com/1.1'.
upload_url (str, optional):
The base URL to use for uploads. Defaults to 'https://upload.twitter.com/1.1'.
chunk_size (int, optional):
Chunk size to use for chunked (multi-part) uploads of images/videos/gifs.
Defaults to 1MB. Anything under 16KB and you run the risk of erroring out
on 15MB files.
use_gzip_compression (bool, optional):
Set to True to tell enable gzip compression for any call
made to Twitter. Defaults to False.
debugHTTP (bool, optional):
Set to True to enable debug output from urllib2 when performing
any HTTP requests. Defaults to False.
timeout (int, optional):
Set timeout (in seconds) of the http/https requests. If None the
requests lib default will be used. Defaults to None.
sleep_on_rate_limit (bool, optional):
Whether to sleep an appropriate amount of time if a rate limit is hit for
an endpoint.
tweet_mode (str, optional):
Whether to use the new (as of Sept. 2016) extended tweet mode. See docs for
details. Choices are ['compatibility', 'extended'].
proxies (dict, optional):
A dictionary of proxies for the request to pass through, if not specified
allows requests lib to use environmental variables for proxy if any.
"""
# check to see if the library is running on a Google App Engine instance
# see GAE.rst for more information
if os.environ:
if 'APPENGINE_RUNTIME' in os.environ.keys():
import requests_toolbelt.adapters.appengine # Adapter ensures requests use app engine's urlfetch
requests_toolbelt.adapters.appengine.monkeypatch()
cache = None # App Engine does not like this caching strategy, disable caching
self.SetCache(cache)
self._cache_timeout = Api.DEFAULT_CACHE_TIMEOUT
self._input_encoding = input_encoding
self._use_gzip = use_gzip_compression
self._debugHTTP = debugHTTP
self._shortlink_size = 19
self._timeout = timeout
self.__auth = None
self._InitializeRequestHeaders(request_headers)
self._InitializeUserAgent()
self._InitializeDefaultParameters()
self.rate_limit = RateLimit()
self.sleep_on_rate_limit = sleep_on_rate_limit
self.tweet_mode = tweet_mode
self.proxies = proxies
if base_url is None:
self.base_url = 'https://api.twitter.com/1.1'
else:
self.base_url = base_url
if stream_url is None:
self.stream_url = 'https://stream.twitter.com/1.1'
else:
self.stream_url = stream_url
if upload_url is None:
self.upload_url = 'https://upload.twitter.com/1.1'
else:
self.upload_url = upload_url
self.chunk_size = chunk_size
if self.chunk_size < 1024 * 16:
warnings.warn((
"A chunk size lower than 16384 may result in too many "
"requests to the Twitter API when uploading videos. You are "
"strongly advised to increase it above 16384"
))
if (consumer_key and not
(application_only_auth or all([access_token_key, access_token_secret]))):
raise TwitterError({'message': "Missing oAuth Consumer Key or Access Token"})
self.SetCredentials(consumer_key, consumer_secret, access_token_key, access_token_secret,
application_only_auth)
if debugHTTP:
import logging
import http.client
http.client.HTTPConnection.debuglevel = 1
logging.basicConfig() # you need to initialize logging, otherwise you will not see anything from requests
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
[docs] def GetAppOnlyAuthToken(self, consumer_key, consumer_secret):
"""
Generate a Bearer Token from consumer_key and consumer_secret
"""
from urllib import quote_plus
import base64
key = quote_plus(consumer_key)
secret = quote_plus(consumer_secret)
bearer_token = base64.b64encode('{}:{}'.format(key, secret))
post_headers = {
'Authorization': 'Basic ' + bearer_token,
'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
}
res = requests.post(url='https://api.twitter.com/oauth2/token',
data={'grant_type': 'client_credentials'},
headers=post_headers)
bearer_creds = res.json()
return bearer_creds
[docs] def SetCredentials(self,
consumer_key,
consumer_secret,
access_token_key=None,
access_token_secret=None,
application_only_auth=False):
"""Set the consumer_key and consumer_secret for this instance
Args:
consumer_key:
The consumer_key of the twitter account.
consumer_secret:
The consumer_secret for the twitter account.
access_token_key:
The oAuth access token key value you retrieved
from running get_access_token.py.
access_token_secret:
The oAuth access token's secret, also retrieved
from the get_access_token.py run.
application_only_auth:
Whether to generate a bearer token and use Application-Only Auth
"""
self._consumer_key = consumer_key
self._consumer_secret = consumer_secret
self._access_token_key = access_token_key
self._access_token_secret = access_token_secret
if application_only_auth:
self._bearer_token = self.GetAppOnlyAuthToken(consumer_key, consumer_secret)
self.__auth = OAuth2(token=self._bearer_token