如何使用 REST API 来做一个命令行 TODO 程序

AVOS Cloud RESTFul API

AVOS Cloud 除了提供几种设备的 SDK 让您可以进行移动 APP 的开发以外, 同样准备了一套 RESTFul API 接口可以供任何可以使用 HTTP 协议的设备和程序来连接 AVOS Cloud 并与之交互, 这样您可以自行开发使用 AVOS Cloud 的程序, 而且可以拥有 AVOS Cloud 的全部功能.

这篇文章中我们就尝试使用 REST API 用 Python 来与 AVOS Cloud 进行交互, 最终做出一个可以使用的 TODO 小程序.

开工前的准备

既然我们要做一个 APP, 那么首先要确定的是什么呢? 自然是确定这个程序到底要做什么事情, 对于 TODO 这种小程序来说, 我们的需求其实很简单:

  1. 可以用用户名和密码登陆, 登陆之后显示当前用户的 todo list
  2. 可以增加 TODO
  3. 可以将 TODO 状态改为完成
  4. 可以删除 TODO
  5. 所有的操作都随时与 AVOS Cloud 同步, 在任何可以运行代码的地方用户都能看到自己的 TODO list

当然这篇教程之后您可以在此基础上任意添加您想要的功能, 您可以随意使用教程中的源码.

注册 AVOS Cloud

既然需要 AVOS Cloud 来提供用户名密码等登陆方式, 自然需要在 cn.avoscloud.com 上面注册一个账户, 我们需要新建一个 APP 来存放我们的数据, 还需要在 APP 中新建一个 Todo 类来存放用户的 todo 信息. 这些都不需要写代码, 直接在网站上操作就可以了.

为了我们的程序能和 AVOS Cloud 服务器连接上, 我们需要 2 个信息, 一个是 Application Id, 另一个是 Application Key, 有了这两个信息以后将来我们就可以轻松地连接 AVOS Cloud 查看和操作我们的数据了.

连接 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 的结构即可.

与 AVOS Cloud 交互

增加/修改/删除 TODO

这些操作在 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

对于我们当前用户来说, 我们要获取 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 程序还是非常简陋的, 它甚至没有安全性, 任何用户都可以访问别人的 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()