使用 Python 和 Flask 实现 RESTful services

使用 Flask 构建 web services 是十分简单地,比我在 Mega-Tutorial 中构建的完整的服务端的应用程序要简单地多。

在 Flask 中有许多扩展来帮助我们构建 RESTful services,但是在我看来这个任务十分简单,没有必要使用 Flask 扩展。

我们 web service 的客户端需要添加、删除以及修改任务的服务,因此显然我们需要一种方式来存储任务。最直接的方式就是建立一个小型的数据库,但是数据库并不是本文的主体。学习在 Flask 中使用合适的数据库,我强烈建议阅读 Mega-Tutorial

这里我们直接把任务列表存储在内存中,因此这些任务列表只会在 web 服务器运行中工作,在结束的时候就失效。 这种方式只是适用我们自己开发的 web 服务器,不适用于生产环境的 web 服务器, 这种情况一个合适的数据库的搭建是必须的。

我们现在来实现 web service 的第一个入口:

  1. #!flask/bin/python
  2. from flask import Flask, jsonify
  3.  
  4. app = Flask(__name__)
  5.  
  6. tasks = [
  7. {
  8. 'id': 1,
  9. 'title': u'Buy groceries',
  10. 'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
  11. 'done': False
  12. },
  13. {
  14. 'id': 2,
  15. 'title': u'Learn Python',
  16. 'description': u'Need to find a good Python tutorial on the web',
  17. 'done': False
  18. }
  19. ]
  20.  
  21. @app.route('/todo/api/v1.0/tasks', methods=['GET'])
  22. def get_tasks():
  23. return jsonify({'tasks': tasks})
  24.  
  25. if __name__ == '__main__':
  26. app.run(debug=True)

正如你所见,没有多大的变化。我们创建一个任务的内存数据库,这里无非就是一个字典和数组。数组中的每一个元素都具有上述定义的任务的属性。

取代了首页,我们现在拥有一个 get_tasks 的函数,访问的 URI 为 /todo/api/v1.0/tasks,并且只允许 GET 的 HTTP 方法。

这个函数的响应不是文本,我们使用 JSON 数据格式来响应,Flask 的 jsonify 函数从我们的数据结构中生成。

使用网页浏览器来测试我们的 web service 不是一个最好的注意,因为网页浏览器上不能轻易地模拟所有的 HTTP 请求的方法。相反,我们会使用 curl。如果你还没有安装 curl 的话,请立即安装它。

通过执行 app.py,启动 web service。接着打开一个新的控制台窗口,运行以下命令:

  1. $ curl -i http://localhost:5000/todo/api/v1.0/tasks
  2. HTTP/1.0 200 OK
  3. Content-Type: application/json
  4. Content-Length: 294
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 04:53:53 GMT
  7.  
  8. {
  9. "tasks": [
  10. {
  11. "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
  12. "done": false,
  13. "id": 1,
  14. "title": "Buy groceries"
  15. },
  16. {
  17. "description": "Need to find a good Python tutorial on the web",
  18. "done": false,
  19. "id": 2,
  20. "title": "Learn Python"
  21. }
  22. ]
  23. }

我们已经成功地调用我们的 RESTful service 的一个函数!

现在我们开始编写 GET 方法请求我们的任务资源的第二个版本。这是一个用来返回单独一个任务的函数:

  1. from flask import abort
  2.  
  3. @app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['GET'])
  4. def get_task(task_id):
  5. task = filter(lambda t: t['id'] == task_id, tasks)
  6. if len(task) == 0:
  7. abort(404)
  8. return jsonify({'task': task[0]})

第二个函数有些意思。这里我们得到了 URL 中任务的 id,接着 Flask 把它转换成 函数中的 task_id 的参数。

我们用这个参数来搜索我们的任务数组。如果我们的数据库中不存在搜索的 id,我们将会返回一个类似 404 的错误,根据 HTTP 规范的意思是 “资源未找到”。

如果我们找到相应的任务,那么我们只需将它用 jsonify 打包成 JSON 格式并将其发送作为响应,就像我们以前那样处理整个任务集合。

调用 curl 请求的结果如下:

  1. $ curl -i http://localhost:5000/todo/api/v1.0/tasks/2
  2. HTTP/1.0 200 OK
  3. Content-Type: application/json
  4. Content-Length: 151
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 05:21:50 GMT
  7.  
  8. {
  9. "task": {
  10. "description": "Need to find a good Python tutorial on the web",
  11. "done": false,
  12. "id": 2,
  13. "title": "Learn Python"
  14. }
  15. }
  16. $ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
  17. HTTP/1.0 404 NOT FOUND
  18. Content-Type: text/html
  19. Content-Length: 238
  20. Server: Werkzeug/0.8.3 Python/2.7.3
  21. Date: Mon, 20 May 2013 05:21:52 GMT
  22.  
  23. <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
  24. <title>404 Not Found</title>
  25. <h1>Not Found</h1>
  26. <p>The requested URL was not found on the server.</p><p>If you entered the URL manually please check your spelling and try again.</p>

当我们请求 id #2 的资源时候,我们获取到了,但是当我们请求 #3 的时候返回了 404 错误。有关错误奇怪的是返回的是 HTML 信息而不是 JSON,这是因为 Flask 按照默认方式生成 404 响应。由于这是一个 Web service 客户端希望我们总是以 JSON 格式回应,所以我们需要改善我们的 404 错误处理程序:

  1. from flask import make_response
  2.  
  3. @app.errorhandler(404)
  4. def not_found(error):
  5. return make_response(jsonify({'error': 'Not found'}), 404)

我们会得到一个友好的错误提示:

  1. $ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
  2. HTTP/1.0 404 NOT FOUND
  3. Content-Type: application/json
  4. Content-Length: 26
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 05:36:54 GMT
  7.  
  8. {
  9. "error": "Not found"
  10. }

接下来就是 POST 方法,我们用来在我们的任务数据库中插入一个新的任务:

  1. from flask import request
  2.  
  3. @app.route('/todo/api/v1.0/tasks', methods=['POST'])
  4. def create_task():
  5. if not request.json or not 'title' in request.json:
  6. abort(400)
  7. task = {
  8. 'id': tasks[-1]['id'] + 1,
  9. 'title': request.json['title'],
  10. 'description': request.json.get('description', ""),
  11. 'done': False
  12. }
  13. tasks.append(task)
  14. return jsonify({'task': task}), 201

添加一个新的任务也是相当容易地。只有当请求以 JSON 格式形式,request.json 才会有请求的数据。如果没有数据,或者存在数据但是缺少 title 项,我们将会返回 400,这是表示请求无效。

接着我们会创建一个新的任务字典,使用最后一个任务的 id + 1 作为该任务的 id。我们允许 description 字段缺失,并且假设 done 字段设置成 False。

我们把新的任务添加到我们的任务数组中,并且把新添加的任务和状态 201 响应给客户端。

使用如下的 curl 命令来测试这个新的函数:

  1. $ curl -i -H "Content-Type: application/json" -X POST -d '{"title":"Read a book"}' http://localhost:5000/todo/api/v1.0/tasks
  2. HTTP/1.0 201 Created
  3. Content-Type: application/json
  4. Content-Length: 104
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 05:56:21 GMT
  7.  
  8. {
  9. "task": {
  10. "description": "",
  11. "done": false,
  12. "id": 3,
  13. "title": "Read a book"
  14. }
  15. }

注意:如果你在 Windows 上并且运行 Cygwin 版本的 curl,上面的命令不会有任何问题。然而,如果你使用原生的 curl,命令会有些不同:

  1. curl -i -H "Content-Type: application/json" -X POST -d "{"""title""":"""Read a book"""}" http://localhost:5000/todo/api/v1.0/tasks

当然在完成这个请求后,我们可以得到任务的更新列表:

  1. $ curl -i http://localhost:5000/todo/api/v1.0/tasks
  2. HTTP/1.0 200 OK
  3. Content-Type: application/json
  4. Content-Length: 423
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 05:57:44 GMT
  7.  
  8. {
  9. "tasks": [
  10. {
  11. "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
  12. "done": false,
  13. "id": 1,
  14. "title": "Buy groceries"
  15. },
  16. {
  17. "description": "Need to find a good Python tutorial on the web",
  18. "done": false,
  19. "id": 2,
  20. "title": "Learn Python"
  21. },
  22. {
  23. "description": "",
  24. "done": false,
  25. "id": 3,
  26. "title": "Read a book"
  27. }
  28. ]
  29. }

剩下的两个函数如下所示:

  1. @app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
    def update_task(task_id):
    task = filter(lambda t: t['id'] == task_id, tasks)
    if len(task) == 0:
    abort(404)
    if not request.json:
    abort(400)
    if 'title' in request.json and type(request.json['title']) != unicode:
    abort(400)
    if 'description' in request.json and type(request.json['description']) is not unicode:
    abort(400)
    if 'done' in request.json and type(request.json['done']) is not bool:
    abort(400)
    task[0]['title'] = request.json.get('title', task[0]['title'])
    task[0]['description'] = request.json.get('description', task[0]['description'])
    task[0]['done'] = request.json.get('done', task[0]['done'])
    return jsonify({'task': task[0]})

  2. @app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['DELETE'])
    def delete_task(task_id):
    task = filter(lambda t: t['id'] == task_id, tasks)
    if len(task) == 0:
    abort(404)
    tasks.remove(task[0])
    return jsonify({'result': True})

delete_task 函数没有什么特别的。对于 update_task 函数,我们需要严格地检查输入的参数以防止可能的问题。我们需要确保在我们把它更新到数据库之前,任何客户端提供我们的是预期的格式。

更新任务 #2 的函数调用如下所示:

  1. $ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://localhost:5000/todo/api/v1.0/tasks/2
  2. HTTP/1.0 200 OK
  3. Content-Type: application/json
  4. Content-Length: 170
  5. Server: Werkzeug/0.8.3 Python/2.7.3
  6. Date: Mon, 20 May 2013 07:10:16 GMT
  7.  
  8. {
  9. "task": [
  10. {
  11. "description": "Need to find a good Python tutorial on the web",
  12. "done": true,
  13. "id": 2,
  14. "title": "Learn Python"
  15. }
  16. ]
  17. }