Last active
January 7, 2016 21:22
-
-
Save barberj/3333993 to your computer and use it in GitHub Desktop.
oauth examples using rauth
This file contains 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
from flask import Flask, request, redirect, url_for, session | |
from rauth.service import OAuth1Service | |
from webbrowser import open | |
import configit | |
config = configit.conf_from_file('development.py') | |
domain = config.magento.domain | |
magento = OAuth1Service( | |
name='magento', | |
consumer_key=config.magento.key, | |
consumer_secret=config.magento.secret, | |
request_token_url='{}/oauth/initiate'.format(domain), | |
access_token_url='{}/oauth/token'.format(domain), | |
authorize_url='{}/admin/oauth_authorize'.format(domain), | |
base_url = '{}/api/rest/'.format(domain)) | |
request_token, request_token_secret = magento.get_request_token( | |
params={'oauth_callback':'oob'}) | |
authorize_url = magento.get_authorize_url(request_token) | |
print 'Visit this URL in your browser: ' + authorize_url | |
open(authorize_url) | |
verifier = raw_input('Enter oauth_verifier from browser: ') | |
session = magento.get_auth_session(request_token, | |
request_token_secret, | |
method='POST', | |
data={'oauth_verifier': verifier}) | |
resp = session.get('customers') | |
import ipdb;ipdb.set_trace() |
This file contains 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
from flask import Flask, request, redirect, url_for, session | |
from rauth.service import OAuth1Service | |
from webbrowser import open | |
import configit | |
config = configit.conf_from_file('development.py') | |
twitter = OAuth1Service( | |
name='twitter', | |
consumer_key=config.twitter.key, | |
consumer_secret=config.twitter.secret, | |
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_url = 'https://api.twitter.com/1/') | |
request_token, request_token_secret = twitter.get_request_token() | |
authorize_url = twitter.get_authorize_url(request_token) | |
print 'Visit this URL in your browser: ' + authorize_url | |
open(authorize_url) | |
verifier = raw_input('Enter oauth_verifier from browser: ') | |
session = twitter.get_auth_session(request_token, | |
request_token_secret, | |
method='POST', | |
data={'oauth_verifier': verifier}) | |
params = {'include_rts': 1, # Include retweets | |
'count': 10} # 10 tweets | |
resp = session.get('statuses/home_timeline.json', params=params) | |
for i, tweet in enumerate(resp.json(), 1): | |
handle = tweet['user']['screen_name'].encode('utf-8') | |
text = tweet['text'].encode('utf-8') | |
print '{0}. @{1} - {2}'.format(i, handle, text) |
This file contains 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
""" | |
Demonstrate magento oauth as a web app. | |
Be sure to update with your consumer key/secret and your callback. | |
""" | |
from flask import Flask, request, redirect, url_for, session | |
from rauth.service import OAuth1Service, OAuth1Session | |
import configit | |
config = configit.conf_from_file('development.py') | |
domain = config.magento.domain | |
magento = OAuth1Service( | |
name='magento', | |
consumer_key=config.magento.key, | |
consumer_secret=config.magento.secret, | |
request_token_url='{}/oauth/initiate'.format(domain), | |
access_token_url='{}/oauth/token'.format(domain), | |
authorize_url='{}/admin/oauth_authorize'.format(domain), | |
base_url = '{}/api/rest/'.format(domain)) | |
app = Flask(__name__) | |
app.debug = True | |
app.secret_key = 'SSssh its a secret' | |
@app.route('/get_access') | |
def access(): | |
app.logger.debug('get magento access tokens\n' | |
'{}\n{}\n'.format(session.get('requst_token', None), | |
session.get('request_token_secret', | |
None))) | |
token = request.args.get('oauth_token', None) | |
verifier = request.args.get('oauth_verifier', None) | |
if session.get('request_token') and (token and verifier): | |
print 'TOKEN: ', token | |
print 'VERIFIER: ', verifier | |
magento_session = magento.get_auth_session( | |
token, | |
request_token_secret=session.get('request_token_secret'), | |
data={'oauth_verifier': verifier} | |
) | |
session.pop('request_token_secret', None) | |
session.pop('request_token', None) | |
session['access_token'] = magento_session.access_token | |
session['access_token_secret'] = magento_session.access_token_secret | |
print session['access_token'], session['access_token_secret'] | |
return redirect(url_for('customers')) | |
else: | |
app.logger.debug('redirecting to authorize') | |
return redirect(url_for('authorize')) | |
@app.route('/') | |
def customers(): | |
app.logger.debug('display magento customers using:\n' | |
'{}\n{}\n'.format(session.get('access_token', None), | |
session.get('access_token_secret', None))) | |
session.pop('access_token', None) | |
session.pop('access_token_secret', None) | |
if not session.get('access_token'): | |
app.logger.debug('redirecting for authorize') | |
return redirect(url_for('authorize')) | |
magento = OAuth1Session( | |
consumer_key=config.magento.key, | |
consumer_secret=config.magento.secret, | |
access_token=session['access_token'], | |
access_token_secret=session['access_token_secret']) | |
response = magento.get('http://magento.kevy.com/api/rest/customers') | |
if response.status_code == 401: | |
session.pop('access_token', None) | |
session.pop('access_token_secret', None) | |
return redirect(url_for('authorize')) | |
import ipdb;ipdb.set_trace() | |
@app.route('/authorize') | |
def authorize(): | |
app.logger.debug('authorize app in magento') | |
""" | |
Request tokens as a desktop app | |
session['request_token'], session['request_token_sercret'] = | |
magento.get_request_token(method='GET') | |
""" | |
session['request_token'], session['request_token_sercret'] =\ | |
magento.get_request_token( | |
params={'oauth_callback':'http://local.kevy.com/get_access'}) | |
authorize_url = magento.get_authorize_url(session['request_token']) | |
app.logger.debug('Authorize url {}'.format(authorize_url)) | |
return redirect(authorize_url) | |
if __name__ == '__main__': | |
app.run(host='0.0.0.0', port=3000) |
This file contains 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
""" | |
Demonstrate twitter oauth as a web app. | |
Be sure to update with your consumer key/secret and your callback. | |
""" | |
from flask import Flask, request, redirect, url_for, session | |
from rauth.service import OAuth1Service, OAuth1Session | |
import configit | |
config = configit.conf_from_file('development.py') | |
twitter = OAuth1Service( | |
name='twitter', | |
consumer_key=config.twitter.key, | |
consumer_secret=config.twitter.secret, | |
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_url = 'https://api.twitter.com/1/') | |
app = Flask(__name__) | |
app.debug = True | |
app.secret_key = 'SSssh its a secret' | |
@app.route('/get_access') | |
def access(): | |
app.logger.debug('get twitter access tokens\n' | |
'{}\n{}\n'.format(session.get('requst_token', None), | |
session.get('request_token_secret', | |
None))) | |
token = request.args.get('oauth_token', None) | |
verifier = request.args.get('oauth_verifier', None) | |
if session.get('request_token') and (token and verifier): | |
print 'TOKEN: ', token | |
print 'VERIFIER: ', verifier | |
twitter_session = twitter.get_auth_session( | |
token, | |
request_token_secret=session.get('request_token_secret'), | |
data={'oauth_verifier': verifier} | |
) | |
session.pop('request_token_secret', None) | |
session.pop('request_token', None) | |
session['access_token'] = twitter_session.access_token | |
session['access_token_secret'] = twitter_session.access_token_secret | |
print session['access_token'], session['access_token_secret'] | |
return redirect(url_for('timeline')) | |
else: | |
app.logger.debug('redirecting to authorize') | |
return redirect(url_for('authorize')) | |
@app.route('/') | |
def timeline(): | |
app.logger.debug('display twitter time line:\n' | |
'{}\n{}\n'.format(session.get('access_token', None), | |
session.get('access_token_secret', None))) | |
if not session.get('access_token'): | |
app.logger.debug('redirecting for authorize') | |
return redirect(url_for('authorize')) | |
params = {'include_rts': 1, # Include retweets | |
'count': 10} # 10 tweets | |
twitter = OAuth1Session( | |
consumer_key=config.twitter.key, | |
consumer_secret=config.twitter.secret, | |
access_token=session['access_token'], | |
access_token_secret=session['access_token_secret']) | |
response = twitter.get('https://api.twitter.com/1/statuses/home_timeline.json', | |
params=params) | |
tweets = '' | |
if response.status_code == 401: | |
session.pop('access_token', None) | |
session.pop('access_token_secret', None) | |
return redirect(url_for('authorize')) | |
assert response.status_code == 200 | |
for i, tweet in enumerate(response.json(), 1): | |
handle = tweet['user']['screen_name'].encode('utf-8') | |
text = tweet['text'].encode('utf-8') | |
tweets += '{0}. @{1} - {2}</br>'.format(i, handle, text) | |
return tweets | |
@app.route('/authorize') | |
def authorize(): | |
app.logger.debug('authorize app in twitter') | |
""" | |
Request tokens as a desktop app | |
session['request_token'], session['request_token_sercret'] = | |
twitter.get_request_token(method='GET') | |
""" | |
session['request_token'], session['request_token_sercret'] =\ | |
twitter.get_request_token( | |
params={'oauth_callback':'http://{}:5000/get_access'.format(config.myip)}) | |
authorize_url = twitter.get_authorize_url(session['request_token']) | |
return redirect(authorize_url) | |
if __name__ == '__main__': | |
app.run(host='0.0.0.0', port=5000) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment