Last active
September 26, 2015 23:08
-
-
Save jcarbaugh/1173914 to your computer and use it in GitHub Desktop.
Very simple, basic access to the Twitter API
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
A very simple client for basic access to the Twitter API. | |
import tweets | |
client = tweets.TwitterClient(CONSUMER_KEY, CONSUMER_SECRET) | |
client.get_access_token() # complete authentication and enter PIN | |
client.get('statuses/home_timeline') | |
Requirements: | |
oauth2 | |
""" | |
from urllib2 import urlopen | |
import json | |
import time | |
import urlparse | |
import webbrowser | |
import oauth2 as oauth | |
REQUEST_TOKEN_URL = 'https://api.twitter.com/oauth/request_token' | |
ACCESS_TOKEN_URL = 'https://api.twitter.com/oauth/access_token' | |
AUTHORIZE_URL = 'https://api.twitter.com/oauth/authorize' | |
BASE_ENDPOINT = "https://api.twitter.com/1.1/" | |
class TwitterClient(): | |
""" Provides very basic access to the Twitter API. | |
By calling get_access_token, you will be taken to Twitter for authorization | |
and prompted to enter the Twitter-provided PIN. | |
""" | |
def __init__(self, consumer_key, consumer_secret): | |
self.consumer = oauth.Consumer(key=consumer_key, secret=consumer_secret) | |
self.sig_method = oauth.SignatureMethod_HMAC_SHA1() | |
self.access_token = None | |
self.response_format = 'json' | |
def get_access_token(self): | |
""" Get an access token from Twitter. | |
Calling this method will open a web browser for authentication on | |
twitter.com. You will be prompted for a PIN that Twitter will return | |
on successful authentication. | |
""" | |
client = oauth.Client(self.consumer) | |
resp, content = client.request(REQUEST_TOKEN_URL, "GET") | |
request_token = dict(urlparse.parse_qsl(content)) | |
webbrowser.open_new("%s?oauth_token=%s" % (AUTHORIZE_URL, request_token['oauth_token'])) | |
oauth_verifier = raw_input('What is the PIN? ') | |
token = oauth.Token(request_token['oauth_token'], request_token['oauth_token_secret']) | |
token.set_verifier(oauth_verifier) | |
client = oauth.Client(self.consumer, token) | |
resp, content = client.request(ACCESS_TOKEN_URL, "POST") | |
access_token = dict(urlparse.parse_qsl(content)) | |
if 'oauth_token' not in access_token: | |
raise oauth.Error("invalid PIN") | |
token = (access_token['oauth_token'], access_token['oauth_token_secret']) | |
self.set_access_token(*token) # set access token on client | |
return token # also return to user so that access token can be saved | |
def set_access_token(self, access_token, access_secret): | |
self.access_token = oauth.Token(access_token, access_secret) | |
def _invoke(self, method, path, params=None): | |
if self.access_token is None: | |
raise ValueError('access token is required to make API call') | |
if method not in ('GET', 'POST'): | |
raise ValueError('only GET and POST methods are supported') | |
params = params or {} | |
params.update({ | |
'oauth_version': "1.0", | |
'oauth_nonce': oauth.generate_nonce(), | |
'oauth_timestamp': int(time.time()), | |
}) | |
url = "%s%s.%s" % (BASE_ENDPOINT, path, self.response_format) | |
request = oauth.Request(method=method, url=url, parameters=params) | |
request.sign_request(self.sig_method, self.consumer, self.access_token) | |
if method == 'POST': | |
resp = urlopen(url, request.to_postdata()) | |
else: | |
resp = urlopen(request.to_url()) | |
resp_content = resp.read() | |
resp.close() | |
return json.loads(resp_content) | |
def get(self, path, params=None): | |
""" Make a GET HTTP request. | |
""" | |
return self._invoke('GET', path, params) | |
def post(self, path, params=None): | |
""" Make a POST HTTP request. | |
""" | |
return self._invoke('POST', path, params) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment