订阅在GAE上用Google日历API发短信
2010 11 20 10:29 PM 4725次查看
分类：Google App Engine 标签：Google App Engine, Google Calendar, Python
《在 App Engine 上使用 Google 数据 API》
《Authentication in the Google Data Protocol》
《AuthSub in the Google Data Protocol Client Libraries》
简单来说，要用Google Data API分为2步：验证和访问资源。
验证有ClientLogin、AuthSub和OAuth这3种方式，其中最简单的就是ClientLogin，只要输入用户名和密码即可：
calendar_client = gdata.calendar.service.CalendarService()
calendar_client.email='你的Google账号用户名'
calendar_client.password='你的Google账号密码'
calendar_client.ProgrammaticLogin()不过如果要对用户开放的话，自然不能使用这种方式，毕竟很不安全，所以我仍然去研究了下推荐的AuthSub验证方式。其实和Twitter API的OAuth授权差不多，我就懒得去研究2者的区别了。
它的步骤就是将用户重定向到Google，用户对其授权后，带着token回到你的应用。这个token是只能使用一次的，而且限制了作用域（scope，其实就是一个URL，标识了可用的服务和路径）。你的应用拿到这个token后，一般会再次访问Google去交换一个session token，这个token是长期可用的，直到用户解除授权。
有了token后，你就可以在HTTP头里加上这行来请求资源了：
Authorization: AuthSub token="yourAuthToken"
当然，这种麻烦的活自然不用我们自己去操心，Google Data APIs Python Client Library里已经自动帮我们做了这些事了。
于是看一个简单的例子：
class Page(webapp.RequestHandler):
def get(self):
self.calendar_client = gdata.calendar.service.CalendarService()
gdata.alt.appengine.run_on_appengine(self.calendar_client)
auth_token = gdata.auth.extract_auth_sub_token_from_url(self.request.uri)
if auth_token:
self.calendar_client.UpgradeToSessionToken(auth_token)
else:
token_request_url = gdata.auth.generate_auth_sub_url(self.request.uri,
('http://www.google.com/calendar/feeds/',))
self.response.out.write('<a href="%s"/>get token</a>' % token_request_url)如果你登录过的话，你会发现数据库里多了一个TokenCollection类型，其中有个实体就是以你的用户名和token构成的。
不过这个库设计得有点不好，不能获取指定用户的token，只能使用当前登录用户的token，这样我就没法在后台自动运行时使用指定用户的token了。
于是便改造了一下gdata.alt.appengine，写了个gdata_for_gae.py：
# -*- coding: utf-8 -*-
from gdata.alt.appengine import *
class GDataToken(db.Model):
tokens = db.BlobProperty()
def save_auth_tokens(token_dict, user=None):
if user is None:
user = users.get_current_user()
if user:
user = user.email()
if user is None:
return None
pickled_token = pickle.dumps(token_dict)
memcache.set('GDataToken:%s' % user, pickled_token)
return GDataToken(key_name=user, tokens=pickled_token).put()
def load_auth_tokens(user=None):
if user is None:
user = users.get_current_user()
if user:
user = user.email()
if user is None:
return {}
pickled_tokens = memcache.get('GDataToken:%s' % user)
if pickled_tokens:
return pickle.loads(pickled_tokens)
user_tokens = GDataToken.get_by_key_name(user)
if user_tokens:
memcache.set('GDataToken:%s' % user, user_tokens.tokens)
return pickle.loads(user_tokens.tokens)
return {}
class AppEngineTokenStore(atom.token_store.TokenStore):
def __init__(self, user=None):
self.user = user
def add_token(self, token):
tokens = load_auth_tokens(self.user)
if not hasattr(token, 'scopes') or not token.scopes:
return False
for scope in token.scopes:
tokens[str(scope)] = token
key = save_auth_tokens(tokens, self.user)
if key:
return True
return False
def find_token(self, url):
if url is None:
return None
if isinstance(url, (str, unicode)):
url = atom.url.parse_url(url)
tokens = load_auth_tokens(self.user)
if url in tokens:
token = tokens[url]
if token.valid_for_scope(url):
return token
else:
del tokens[url]
save_auth_tokens(tokens, self.user)
for scope, token in tokens.iteritems():
if token.valid_for_scope(url):
return token
return atom.http_interface.GenericToken()
def remove_token(self, token):
token_found = False
scopes_to_delete = []
tokens = load_auth_tokens(self.user)
for scope, stored_token in tokens.iteritems():
if stored_token == token:
scopes_to_delete.append(scope)
token_found = True
for scope in scopes_to_delete:
del tokens[scope]
if token_found:
save_auth_tokens(tokens, self.user)
return token_found
def remove_all_tokens(self):
save_auth_tokens({}, self.user)
def run_on_appengine(gdata_service, store_tokens=True,
single_user_mode=False, deadline=None, user=None):
gdata_service.http_client = AppEngineHttpClient(deadline=deadline)
gdata_service.token_store = AppEngineTokenStore(user)
gdata_service.auto_store_tokens = store_tokens
gdata_service.auto_set_current_token = single_user_mode
return gdata_servicegdata_for_gae.run_on_appengine(self.calendar_client) # 使用当前用户的token
gdata_for_gae.run_on_appengine(self.calendar_client, user="email adderss") # 使用指定用户的token拿到token后，还要和calendar_client关联起来：
其中上面那行run_on_appengine的代码也会自动获取数据库里的token，不过数据库里也不一定有这个用户的token。
此外，self.calendar_client.UpgradeToSessionToken(auth_token)也是一种设置token的方式，其他的基本上都是它的变种了。
calendar_client和token关联完成后，就可以用它访问资源了。
event_entry = gdata.calendar.CalendarEventEntry()
event_entry.title = atom.Title(text='test')
event_entry.content = atom.Content(text='test')
start_time = time.strftime('%Y-%m-%dT%H:%M:%S.000Z', time.gmtime(time.time() + 80))
event_entry.when.append(gdata.calendar.When(start_time=start_time), reminder=gdata.calendar.Reminder(minutes=1, method='sms'))
try:
cal_event = self.calendar_client.InsertEvent(event_entry, 'http://www.google.com/calendar/feeds/default/private/full')
except:
pass你也可以使用其他的日历feed，方法就是创建或选择一个你拥有的日历，点下右侧那个倒三角，选择“日历设置”。在这个设置页面中，往下找到“私人网址”，其中XML图标对应的就是这个日历的feed了。
不过这个feed是只读的，它的格式类似于：http://www.google.com/calendar/feeds/.....%40group.calendar.google.com/private-...../basic
把“private-...../basic”改成“private/full”后就是可写的feed地址，即：http://www.google.com/calendar/feeds/.....%40group.calendar.google.com/private/full
此外，如果要用自己的域名的话（非appspot.com域名），需要在Google的Manage Your Domains页面进行注册。详细方法可见Registration for Web-Based Applications。
注意我把提醒时间设为了提前1分钟，发生时间为80秒后，也就是大约20秒后你就会收到Google发来的短信了。
废话就不再说了，我去给Doodel加短信提醒功能去＝。＝
向下滚动可载入更多评论，或者点这里禁止自动加载。