AVOS Cloud除了提供几种设备的SDK让您可以进行移动APP的开发以外,同样准备了一套RESTFul API接口可以供任何可以使用HTTP协议的设备和程序来连接AVOS Cloud并与之交互,这样您可以自行开发使用AVOS Cloud的程序,而且可以拥有AVOS Cloud的全部功能.
这篇文章中我们就尝试使用REST API用Python来与AVOS Cloud进行交互,最终做出一个可以使用的TODO小程序.
既然我们要做一个APP,那么首先要确定的是什么呢?自然是确定这个程序到底要做什么事情,对于TODO这种小程序来说,我们的需求其实很简单:
当然这篇教程之后您可以在此基础上任意添加您想要的功能,您可以随意使用教程中的源码.
既然需要AVOS Cloud来提供用户名密码等登陆方式,自然需要在 cn.avoscloud.com 上面注册一个账户,我们需要新建一个APP来存放我们的数据,还需要在APP中新建一个Todo类来存放用户的todo信息.这些都不需要写代码,直接在网站上操作就可以了.
为了我们的程序能和AVOS Cloud服务器连接上,我们需要2个信息,一个是Application Id,另一个是Application Key,有了这两个信息以后将来我们就可以轻松地连接AVOS Cloud查看和操作我们的数据了.
AVOS Cloud的REST接口需要我们自己发送不同的HTTP请求来与之交互,在程序中我们用httplib来做到这一点,同时再引入json和urllib来做json和url的编码解码工作.
在 Python中,建立连接的代码是这样的:
connection = httplib.HTTPSConnection('cn.avoscloud.com')
connection.connect()
我们可以用connection这个对象来发出和接受请求,按照RSET API手册上所说的,我们尝试一下注册一个用户,他的用户名是tester,密码是123456:
header = {
"X-AVOSCloud-Application-Id" : "6w2rdw5x7l583ms221q7mchykbam9hr37yxvmhlsprtgxr39",
"X-AVOSCloud-Application-Key" : "pr1o4ux3og6qe9htuvuaklv348b7oqa73uqz30uyhwo3c6hi",
"Content-Type" : "application/json",
}
body_d = {
"username": "tester",
"password": "123456",
}
body = json.dumps(body_d)
connection.request('POST','/1/users',body,header)
注意这里的"X-AVOSCloud-Application-Id"和"X-AVOSCloud-Application-Key"就是我们在注册AVOS Cloud的时候得到的Application Id和Application Key,如果发送的字符串是经过json编码的话需要在header中指定Content-Type.我们通过connection.request方法发送一个请求,接着可以通过connection.getresponse方法得到AVOSCloud返回给我们的结果:
response = connection.getresponse()
result = json.loads(response.read())
这里需要用json.loads来把response返回的json字符串转化为python数据结构.
最后用connection.close()结束这次连接:
connection.close()
上面这3步就是每一次对AVOSCloud做请求所经历的步骤,在附件中提供的代码的restlib.py中已经用一个AVConnection封装了,每一次连接只需要使用with语句即可(不要忘了改Application Id和Application Key):
with AVConnection() as a:
result = a.request(method,rul,body,header)
return result
对于我们之前定义的极为简单的todo程序,如果我们要把数据保存在本地,其实这个程序就非常简单,难度大致上就是比Hello World稍微难一点的水准.在完成本地的基础上,您可以发现实际上把程序改成AVOSCloud版本的话只需要很少的工作就可以.
###显示TODO List
对于每个TODO 都有相应的状态(是否完成),也有相应的序号以供之后进行增加和删除操作用,为了确认用户是否成功操作,应该在每次操作后都显示现有的TODO List,我们假设把todo封装在todolist里面:
def print_todolist(todolist):
if len(todolist) == 0:
print "No todos now!"
for i,todo in enumerate(todolist):
done_q = "DONE" if todo[u"done"] else "TODO"
print "%3d : %s : %s" % (i, done_q, todo[u'content'])
注意每一个我们从AVOS Cloud中取回的对象都是经过UTF-8编码后的对象,所以在Python 2中我们对于这些对象都应使用 u''字符串的形式,这样可以方便的支持多国语言.
###操作循环
我们用最简单的循环方式实现在命令行里面用户的操作:
command = raw_input("Input q to quit, 0 to add , 1 to toggle, 2 to remove:")
while (command != 'q'):
if command == "0":
ctr_add_todo(user_objectId)
elif command == "1":
ctr_toggle_todo(todolist)
elif command == "2":
ctr_remove_todo(todolist)
else:
print "Not correct command!"
todolist = retrive_todolist(user_objectId)
print_todolist(todolist)
command = raw_input("Input q to quit, 0 to add , 1 to toggle, 2 to remove:")
其中ctr_add_todo等函数涉及与AVOS Cloud的交互,user_objectId保存的当前用户的用户id.这些在下面都有解释,这里只用看下整个while的结构即可.
这些操作在RESTFul API中均有定义,只需要按照需求发送相应的请求即可,例如我们要修改一个TODO的状态,只需要传入{"done":done}到相应的TODO的objectId上即可更新这个TODO的状态:
def update_obj(clazz,objectId,obj):
body = json.dumps(obj)
with AVConnection() as a:
result = a.json_request('PUT','/1/classes/'+clazz+'/'+objectId,body)
return result
def update_todo(todo_objectId,done):
update_obj("Todo",todo_objectId,{"done":done})
对于增加的TODO来说,我们需要指定这个TODO所隶属的用户,这里需要一个指针指向我们当前用户:
def create_obj(clazz,obj):
body = json.dumps(obj)
with AVConnection() as a:
result = a.json_request('POST','/1/classes/'+clazz,body)
return result
def add_todo(user_objectId,content,done=False):
obj = {
"content" : content,
"user" : {
"__type": "Pointer",
"className": "_User",
"objectId": user_objectId,
},
"done" : done,
}
todo = create_obj("Todo",obj)
对于我们当前用户来说,我们要获取TODO的user字段为当前User,这里就需要我们使用query来获取符合相应条件的内容:
def query_objs(clazz,condition):
params = urllib.urlencode({"where":json.dumps(condition)})
with AVConnection() as a:
result = a.request('GET',('/1/classes/'+clazz+'?%s') % params , '')
return result
def retrive_todolist(user_objectId):
user_condition = {
"user" : {
"__type": "Pointer",
"className": "_User",
"objectId": user_objectId,
}
}
result = query_objs("Todo",user_condition)
if result[1] != 200:
print "Not Found!"
return
todolist = result[0][u"results"]
return todolist
Query有很多高级的用法来查询,您可以查阅RESTFul API来获得更详细的信息.
我们只是用到了很少一些AVOS Cloud的功能就已经能做出一个功能基本成型的云同步的待办事项程序,您可以看到从传统程序出发只需很少的工作量即可让您的程序与云端同步,您可以尽情地发挥您的想象力和创造力来使用AVOS Cloud.
您可以通过任何有网络连接的设备来与AVOS Cloud交互,希望您能
您可以看到这个TODO程序还是非常简陋的,它甚至没有安全性,任何用户都可以访问别人的TODO(如果发送相应的请求的话),要完善这一点的话,您需要在创建这个对象的时候指定ACL,这里留给您自行探索(请查看我们的RESTFul API文档).
##源码
restlib.py
import httplib,json,urllib
class AVConnection:
def __enter__(self):
self.Application_Id = "6w2rdw5x7l583ms221q7mchykbam9hr37yxvmhlsprtgxr39"
self.Application_Key = "pr1o4ux3og6qe9htuvuaklv348b7oqa73uqz30uyhwo3c6hi"
self.connection = httplib.HTTPSConnection('cn.avoscloud.com')
self.connection.connect()
return self
def request(self,method,url,body,header={}):
header["X-AVOSCloud-Application-Id"] = self.Application_Id
header["X-AVOSCloud-Application-Key"] = self.Application_Key
self.connection.request(method,url,body,header)
response = self.connection.getresponse()
result = json.loads(response.read())
status = response.status
return result,status
def json_request(self,method,url,body,header={}):
header["Content-Type"]="application/json"
return self.request(method,url,body,header)
def __exit__(self, type, value, traceback):
self.connection.close()
def login(username,password):
params_d = {
"username":username,
"password":password,
}
params = urllib.urlencode(params_d)
res = {"login":False}
with AVConnection() as a:
result = a.request('GET','/1/login?%s' % params, '',{})
if result[1] == 200:
res.update({"login":True,"body":result[0]})
else:
res["reason"] = result[0][u'error']
return res
def signup(username,password,info={}):
body_d = {
"username": username,
"password": password,
}
body_d.update(info)
body = json.dumps(body_d)
with AVConnection() as a:
result = a.json_request('POST','/1/users',body)
return result
def create_obj(clazz,obj):
body = json.dumps(obj)
with AVConnection() as a:
result = a.json_request('POST','/1/classes/'+clazz,body)
return result
def update_obj(clazz,objectId,obj):
body = json.dumps(obj)
with AVConnection() as a:
result = a.json_request('PUT','/1/classes/'+clazz+'/'+objectId,body)
return result
def get_obj(clazz,objectId):
with AVConnection() as a:
result = a.request('GET','/1/classes/'+clazz+'/'+objectId,'')
return result
def remove_obj(clazz,objectId):
with AVConnection() as a:
result = a.request('DELETE','/1/classes/'+clazz+'/'+objectId,'')
return result
def query_objs(clazz,condition):
params = urllib.urlencode({"where":json.dumps(condition)})
with AVConnection() as a:
result = a.request('GET',('/1/classes/'+clazz+'?%s') % params , '')
return result
todo.py
import restlib,getpass
def retrive_todolist(user_objectId):
user_condition = {
"user" : {
"__type": "Pointer",
"className": "_User",
"objectId": user_objectId,
}
}
result = restlib.query_objs("Todo",user_condition)
if result[1] != 200:
print "Not Found!"
return
todolist = result[0][u"results"]
return todolist
def add_todo(user_objectId,content,done=False):
obj = {
"content" : content,
"user" : {
"__type": "Pointer",
"className": "_User",
"objectId": user_objectId,
},
"done" : done,
}
todo = restlib.create_obj("Todo",obj)
def delete_todo(todo_objectId):
restlib.remove_obj("Todo",todo_objectId)
def update_todo(todo_objectId,done):
restlib.update_obj("Todo",todo_objectId,{"done":done})
def print_todolist(todolist):
if len(todolist) == 0:
print "No todos now!"
for i,todo in enumerate(todolist):
done_q = "DONE" if todo[u"done"] else "TODO"
print "%3d : %s : %s" % (i, done_q, todo[u'content'])
def ctr_add_todo(user_objectId):
content = raw_input("Content:")
add_todo(user_objectId,content)
def ctr_toggle_todo(todolist):
idx = input("Which one?")
todo_objectId = todolist[idx][u'objectId']
done = not todolist[idx][u'done']
update_todo(todo_objectId,done)
def ctr_remove_todo(todolist):
idx = input("Which one?")
todo_objectId = todolist[idx][u'objectId']
delete_todo(todo_objectId)
def main():
username = raw_input("Username:")
password = getpass.getpass("Password:")
result = restlib.login(username,password)
if not result["login"]:
print result["reason"]
return
user_objectId = result["body"][u'objectId']
todolist = retrive_todolist(user_objectId)
print_todolist(todolist)
command = raw_input("Input q to quit, 0 to add , 1 to toggle, 2 to remove:")
while (command != 'q'):
if command == "0":
ctr_add_todo(user_objectId)
elif command == "1":
ctr_toggle_todo(todolist)
elif command == "2":
ctr_remove_todo(todolist)
else:
print "Not correct command!"
todolist = retrive_todolist(user_objectId)
print_todolist(todolist)
command = raw_input("Input q to quit, 0 to add , 1 to toggle, 2 to remove:")
if __name__ == '__main__':
main()