====== Google App Engineで手軽にOAuthアプリを作成！（Twitterとか！） - AppEngine-OAuth ======
更新履歴とコメントは[[http://0-oo.net/log/category/python-box/google-app-engine-twitter/|AppEngine-Twitter Archive - ゼロと無限の間のログ]]へどうぞ。
{{:javascript:09_s.jpg|}}
サイトに対してパスワードを預けないで済むという画期的なプロトコルOAuth。仕様はややこしいようで簡単、なようで意外に複雑。\\
日本語情報が少ないのが痛い。:-\
でも1回作ってしまえば使い回しが効く。ということでライブラリ化した。
今のところTwitter APIでの使用しか考えていないが、他のサービスでもある程度使えるんじゃないかなー。
===== ライセンス =====
[[http://0-oo.net/pryn/MIT_license.txt|MITライセンス]]で。
===== デモ（動作サンプル） =====
==== その1. RT（ReTweet）の多いつぶやきを報告するBot ====
[[http://twitter.com/RT_report|@RT_report]]
==== その2. 地名を告げるとその近くにいるTwitterユーザーの名前を教えてくれるBot ====
[[http://twitter.com/who_is_near|@who_is_near]]
=== 使用例 ===
@who_is_near 六本木
試しにつぶやくには、Twitterにログイン後に[[http://twitter.com/?status=%40who_is_near%20%E5%85%AD%E6%9C%AC%E6%9C%A8|ここをクリック]]
==== その3. 出発地点と目的地を告げると、乗換（ルート）案内のGoogleマップを表示してくれるBot ====
[[http://twitter.com/norikae_map|@norikae_map]]
=== 使用例 1. 移動方法を指定しないと、電車でのルート検索・乗り換え案内になる ===
@norikae_map 東京 大阪
試しにつぶやくには、Twitterにログイン後に[[http://twitter.com/?status=%40norikae_map%20%E6%9D%B1%E4%BA%AC%20%E5%A4%A7%E9%98%AA|ここをクリック]]
=== 使用例 2. 車でのルートの案内 ===
@norikae_map 木更津駅 羽田空港駅 車
試しにつぶやくには、Twitterにログイン後に[[http://twitter.com/?status=%40norikae_map%20%E6%9C%A8%E6%9B%B4%E6%B4%A5%E9%A7%85%20%E7%BE%BD%E7%94%B0%E7%A9%BA%E6%B8%AF%E9%A7%85%20%E8%BB%8A|ここをクリック]]
=== 使用例 3. 歩きでのルートの案内 ===
@norikae_map 霞が関 霞が関 歩き
試しにつぶやくには、Twitterにログイン後に[[http://twitter.com/?status=%40norikae_map%20%E9%9C%9E%E3%81%8C%E9%96%A2%20%E9%9C%9E%E3%81%8C%E9%96%A2%20%E6%AD%A9%E3%81%8D|ここをクリック]]
==== その4. TwitterにOAuthでログインしてTwitter APIを使うデモ ====
[[http://0-oo.appspot.com/oauth/|TwitterにOAuthでログインしてTwitter APIを使うデモ]]
ソースコード\\
（※[[python-box/appengine-basehandler|AppEngine Basehandler]]と[[python-box/appengine-twitter|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 json
import logging
import webapp2
from appengine_twitter import AppEngineTwitter
from basehandler import BaseHandler, h
from google.appengine.ext import db
OAUTH_KEY = 'xxxxx'
OAUTH_SECRET = 'xxxxx'
class DemoHandler(BaseHandler):
def demo_header(self):
self.simple_header(u'AppEngine-OAuthのデモ')
self.p(u'AppEngine-OAuthのデモ
')
def demo_footer(self):
self.p('
')
self.p('
', True)
self.p(u'このページは')
self.p('')
self.p(u'AppEngine-OAuth と ')
self.p('')
self.p(u'AppEngine-Twitter のデモです')
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('')
self.p(u'TwitterのOAuthログインへ', 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 = json.loads(twitter.last_response.content)['screen_name']
self.demo_header()
self.p(u'こんにちは、' + name + u'さん！', True)
self.p('
')
# 以後は再ログインせずに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('
')
if tw2.is_following('uresuji_books') == False:
tw2.follow('uresuji_books')
self.p(u'@uresuji_books をフォローしたよ', True)
self.p('
')
self.p(u'最初のページへ戻る', 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 = webapp2.WSGIApplication(routing, debug=False)
===== AppEngine-OAuthのソースコード =====
#!/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.1'
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