更新履歴とコメントはAppEngine-Twitter Archive - ゼロと無限の間のログへどうぞ。
サイトに対してパスワードを預けないで済むという画期的なプロトコルOAuth。仕様はややこしいようで簡単、なようで意外に複雑。
日本語情報が少ないのが痛い。
でも1回作ってしまえば使い回しが効く。ということでライブラリ化した。
今のところTwitter APIでの使用しか考えていないが、他のサービスでもある程度使えるんじゃないかなー。
@who_is_near 六本木
試しにつぶやくには、Twitterにログイン後にここをクリック
@norikae_map 東京 大阪
試しにつぶやくには、Twitterにログイン後にここをクリック
@norikae_map 木更津駅 羽田空港駅 車
試しにつぶやくには、Twitterにログイン後にここをクリック
@norikae_map 霞が関 霞が関 歩き
試しにつぶやくには、Twitterにログイン後にここをクリック
TwitterにOAuthでログインしてTwitter APIを使うデモ
ソースコード
（※AppEngine BasehandlerとAppEngine Twitterを使っている）
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
Sample for AppEngine-OAuth on Google App Engine
See: http://0-oo.net/sbox/python-box/appengine-oauth
See also: http://apiwiki.twitter.com/OAuth-FAQ
'''
import logging
import wsgiref.handlers
from appengine_twitter import AppEngineTwitter
from basehandler import BaseHandler, h
from django.utils import simplejson
from google.appengine.ext import db
from google.appengine.ext import webapp
OAUTH_KEY = 'xxxxx'
OAUTH_SECRET = 'xxxxx'
class DemoHandler(BaseHandler):
def demo_header(self):
self.simple_header(u'AppEngine-OAuthのデモ')
self.p(u'<h1>AppEngine-OAuthのデモ</h1>')
def demo_footer(self):
self.p('<br>')
self.p('<hr>', True)
self.p(u'このページは')
self.p('<a href="http://0-oo.net/sbox/python-box/appengine-oauth">')
self.p(u'AppEngine-OAuth</a> と ')
self.p('<a href="http://0-oo.net/sbox/python-box/appengine-twitter">')
self.p(u'AppEngine-Twitter</a> のデモです')
self.simple_footer()
class InitHandler(DemoHandler):
def get(self):
twitter = AppEngineTwitter()
twitter.set_oauth(key=OAUTH_KEY, secret=OAUTH_SECRET)
req_info = twitter.prepare_oauth_login()
# request_tokenはcallbackされた後で使うのでDatastoreに保存しておく
OAuthRequestToken(token=req_info['oauth_token'],
secret=req_info['oauth_token_secret']).put()
self.demo_header()
self.p('<a href="' + h(req_info['url']) + '">')
self.p(u'TwitterのOAuthログインへ</a>', True)
self.demo_footer()
class CallbackHandler(DemoHandler):
def get(self):
twitter = AppEngineTwitter()
twitter.set_oauth(OAUTH_KEY, OAUTH_SECRET)
# TwitterからHTTP GETでrequest_tokenが渡される
req_token = self.request.get('oauth_token')
# Datastoreに保存しておいたreqest_token_secretを取り出す
query = OAuthRequestToken.all()
query.filter('token = ', req_token)
req_tokens = query.fetch(1)
# request_tokenとaccess_tokenを交換する
acc_token = twitter.exchange_oauth_tokens(req_token, req_tokens[0].secret)
# ここまで来ればOAuthを使ってAPIが使える。試しにユーザー名を取得
twitter.verify()
name = simplejson.loads(twitter.last_response.content)['screen_name']
self.demo_header()
self.p(u'こんにちは、' + name + u'さん！', True)
self.p('<br>')
# 以後は再ログインせずにaccess_tokenを繰り返し使うことができる
tw2 = AppEngineTwitter()
tw2.set_oauth(OAUTH_KEY,
OAUTH_SECRET,
acc_token['oauth_token'],
acc_token['oauth_token_secret'])
msg = u'AppEngine-OAuth (on Python and Google App Engine) '
msg += u'http://0-oo.appspot.com/oauth/ からこんにちは！'
tw2.update(msg.encode('utf8'))
self.p(u'つぶやいたよ（自分のTLを見てね）', True)
self.p('<br>')
if tw2.is_following('uresuji_books') == False:
tw2.follow('uresuji_books')
self.p(u'@uresuji_books をフォローしたよ', True)
self.p('<br>')
self.p(u'<a href="./">最初のページへ戻る</a>', True)
self.demo_footer()
#Model(s)
class OAuthRequestToken(db.Model):
token = db.StringProperty()
secret = db.StringProperty()
logging.getLogger()
routing = [('/oauth/', InitHandler),
('/oauth/callback', CallbackHandler)]
application = webapp.WSGIApplication(routing, debug=False)
wsgiref.handlers.CGIHandler().run(application)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
AppEngine-OAuth
OAuth utility for applications on Google App Engine
See: http://0-oo.net/sbox/python-box/appengine-oauth
License: http://0-oo.net/pryn/MIT_license.txt (The MIT license)
'''
__author__ = 'dgbadmin@gmail.com'
__version__ = '0.1.0'
import hmac
import urllib
from google.appengine.api import urlfetch
from hashlib import sha1
from random import getrandbits
from time import time
class AppEngineOAuth(object):
def __init__(self, key, secret, acs_token='', acs_token_secret=''):
self._key = key
self._secret = secret
self._token = acs_token
self._token_secret = acs_token_secret
# Be understandable which type token is (request or access)
if acs_token == '':
self._token_type = None
else:
self._token_type = 'access'
def prepare_login(self, req_token_url):
'''
Return request_token, request_token_secret and params of authorize url.
'''
# Get request token
params = self.get_oauth_params(req_token_url, {})
res = urlfetch.fetch(url=req_token_url + '?' + urllib.urlencode(params),
method='GET')
self.last_response = res
if res.status_code != 200:
raise Exception('OAuth Request Token Error: ' + res.content)
# Response content is request_token
dic = self._qs2dict(res.content)
self._token = dic['oauth_token']
self._token_secret = dic['oauth_token_secret']
self._token_type = 'request'
# Get params with signature
sig_params = {'oauth_signature': params['oauth_signature']}
dic['params'] = urllib.urlencode(self.get_oauth_params(req_token_url,
sig_params))
return dic
def exchange_tokens(self, acs_token_url, req_token, req_token_secret):
self._token = req_token
self._token_secret = req_token_secret
self._token_type = 'request'
params = urllib.urlencode(self.get_oauth_params(acs_token_url, {}))
res = urlfetch.fetch(url=acs_token_url, payload=params, method='POST')
self.last_response = res
if res.status_code != 200:
raise Exception('OAuth Access Token Error: ' + res.content)
# Response content is access_token
dic = self._qs2dict(res.content)
self._token = dic['oauth_token']
self._token_secret = dic['oauth_token_secret']
self._token_type = 'access'
return dic
def get_oauth_params(self, url, params, method='GET'):
oauth_params = {'oauth_consumer_key': self._key,
'oauth_signature_method': 'HMAC-SHA1',
'oauth_timestamp': int(time()),
'oauth_nonce': getrandbits(64),
'oauth_version': '1.0'}
if self._token_type != None:
oauth_params['oauth_token'] = self._token
# Add other params
params.update(oauth_params)
# Sort and concat
s = ''
for k in sorted(params):
s += self._quote(k) + '=' + self._quote(params[k]) + '&'
msg = method + '&' + self._quote(url) + '&' + self._quote(s[:-1])
# Maybe token_secret is empty
key = self._secret + '&' + self._token_secret
digest = hmac.new(key, msg, sha1).digest()
params['oauth_signature'] = digest.encode('base64')[:-1]
return params
def _quote(self, s):
return urllib.quote(str(s), '')
def _qs2dict(self, s):
dic = {}
for param in s.split('&'):
(key, value) = param.split('=')
dic[key] = value
return dic
