基于Flask设计RESTful API

Flask是一个Python WEB开发框架。接下来一段闲暇时间打算做的“西子快讯”项目中，我打算拿它来做RESTful API。虽然之前对Python并不了解，对Flask更是陌生，但爱折腾的本性促使我不断去学习。于是今天看blog、看文档，小有收获，略记一二。

准备工作

确保机器上已经安装了Python、Pip、Virtualenv等工具。如果默认的Pip安装源比较慢，可以用v2ex的试试。具体方法是在~/.pip/下创建一个名为pip.conf的文件，输入新的pip源。

[global]
index-url = http://pypi.v2ex.com/simple

安装Flask

为了不影响系统的Python环境，我们用Virtualenv虚拟一个运行环境，然后在此环境下开发一个名为todo的示例小应用（纯RESTful API，JSON格式的数据）。

cd ~/Desktop
mkdir todo-api
cd todo-api
virtualenv venv
. venv/bin/activate

此时命令行窗口应出现(venv)mbpr2013:todo-api linkoubian$ 这样的提示符，注意最左边的venv表明当前运行的是虚拟出来的环境。工作完成后打算退出venv，只需执行deactivate命令即可。

这时执行pip install flask便可在venv下安装flask模块。

测试Flask是否正常工作

编写一个Hello World程序，如下：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
  return "Hello, World!"

if __name__ == '__main__':
  app.run(debug = True)

保存为~/Desktop/todo-api/app.py，添加可执行权限后，敲python app.py开始运行Flask。此时打开浏览器，访问localhost:5000即可看到Hello, World!

开始设计RESTful API

获取所有的tasks

一个task由id、title、description、done等属性构成，获取task列表的url可以设计成/todo/tasks。具体代码如下，

from flask import Flask, jsonify

app = Flask(__name__)

tasks = [
  {
      'id': 1,
      'title': u'Buy groceries',
      'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
      'done': False
  },
  {
      'id': 2,
      'title': u'Learn Python',
      'description': u'Need to find a good Python tutorial on the web',
      'done': False
  },
]

@app.route('/todo/tasks', methods = ['GET'])
def get_tasks():
  return jsonify({'tasks': tasks})

if __name__ == '__main__':
  app.run(debug = True)

因为Flask运行在debug模式下会自动reload，所以代码写完后保存文件即可。

Mac上有一款名为Paw的HTTP Client，算是我见过、用过的各种API测试工具中的佼佼者，力荐。

打开Paw，在url里输入http://localhost:5000/todo/tasks，选择GET，敲CMD+Enter即可听到表示成功的清脆提示音。Paw除了提供原生的HTTP响应格式，还提供格式化后浏览功能，选择JSON格式即可。

上面这段代码还是比较清晰好懂的，首先创建一个task数组，初始化两个task对象放进去。指定GET方式请求/todo/tasks资源时执行get_tasks方法，具体是把前面创建的tasks数组以JSON的格式返回。而将Python的数组转成JSON格式是通过Flask的jsonify函数实现的。

根据id获得特定task

通常我们将id放在URL中，然后对应的处理函数从URL中得到id以从数据源（此处为task数组，存在内存中）中查询相应地entity（此处为task）。

@app.route('/todo/tasks/<int:task_id>', methods = ['GET'])
def get_task(task_id):
  task = filter(lambda t: t['id'] == task_id, tasks)
  if len(task) == 0:
      abort(404)
  return jsonify({'task': task[0]})

为了正常使用abort函数，需要从flask模块import一下。通过Paw测试http://localhost:5000/todo/tasks/1，发现一切正常。

访问http://localhost:5000/todo/tasks/0时，返回一段出错的HTML，这是Flask对404的默认处理，我们可以自己定义一个返回JSON格式的错误消息。

@app.errorhandler(404)
def not_found(error):
    return make_response(jsonify( { 'error': 'Not found' } ), 404)

我们用make_response函数，将构造好的JSON作为404错误的响应，返回给客户端。

创建一个task

URL仍然用/todo/tasks，但HTTP方法为POST。根据Request对象中的数据构造一个新的task对象，插入数据源即可。

@app.route('/todo/tasks', methods = ['POST'])
def create_task():
  if not request.json or not 'title' in request.json:
      abort(400)
  task = {
      'id': tasks[-1]['id'] + 1,
      'title': request.json['title'],
      'description': request.json.get('description', ''),
      'done': False
  }
  tasks.append(task)
  return jsonify({'task': task}), 201

另外，我们需要定义一个400的处理器，类似于404，

@app.errorhandler(400)
def not_found(error):
    return make_response(jsonify( { 'error': 'Bad request' } ), 400)

在Paw中，POST请求http://localhost:5000/todo/tasks，同时Header里加一个Content-Type，值为application/json，Body里加入一段文本内容{“title”: “Buy a cell phone”}，敲CMD+Enter执行。

根据REST Cookbook一书的建议，可以在返回的新增task里加入一个uri字段。为此，建一个名为make_public_task的函数，如下，

def make_public_task(task):
  new_task = {}
  for field in task:
      new_task[field] = task[field]
      if field == 'id':
          new_task['uri'] = url_for('get_task', task_id = task['id'], _external = True)

  return new_task

另外，create_task中改为，

return jsonify({'task': make_public_task(task)}), 201

get_tasks中改为，

return jsonify({'tasks': map(make_public_task, tasks)})

get_task中改为，

return jsonify({'tasks': map(make_public_task, tasks)})

更新一个task

显然此时的HTTP方法为PUT，相应的url和获取task类似。

@app.route('/todo/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': make_public_task(task[0]) } )

删除一个task

@app.route('/todo/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 } )

修改Python源文件后，Flask会reload，这将导致上一次create的task丢失，所以直接尝试delete之前增加的task可能会报404资源找不到错误。

为WEB服务增加安全机制

基于Flask的HTTBasicAuth扩展，定义两个函数如下，

auth = HTTPBasicAuth()

@auth.get_password
def get_password(username):
    if username == 'linkoubian':
        return '7c4a8d09ca3762af61e59520943dc26494f8941b'
    return None

@auth.error_handler
def unauthorized():
    return make_response(jsonify( { 'error': 'Unauthorized access' } ), 401)

其中get_password方法是返回指定用户的密码信息（通常是从数据库用户表查询得到），当此处返回的值和客户端传递的值匹配时（匹配过程由HTTPBasicAuth完成）成功，否则执行error_handler返回错误信息。

定义完校验相关的代码，还需将校验逻辑应用到需要校验的WEB服务。比如，希望只有授权用户才可以修改或删除task，则可以在update_task及delete_task方法上面加上@auth.login_required即可。

此时在Paw中需添加Authorization头，值为HTTP Basic Auth，在username及password文本框中分布输入信息，敲CMD+Enter进行测试。

Paw HTTP Client

说明

本文是根据Miguel Grinberg写的这篇文章翻译并整理的。

Balloonsys Inc.

Just a blog now.