转：使用python的Flask实现一个RESTful API服务器端

提示：可以学习一下flask框架中对于密码进行校验的部分。封装了太多操作。

最近这些年，REST已经成为web services和APIs的标准架构，很多APP的架构基本上是使用RESTful的形式了。

本文将会使用python的Flask框架轻松实现一个RESTful的服务。

REST的六个特性：

• Client-Server：服务器端与客户端分离。
• Stateless（无状态）：每次客户端请求必需包含完整的信息，换句话说，每一次请求都是独立的。
• Cacheable（可缓存）：服务器端必需指定哪些请求是可以缓存的。
• Layered System（分层结构）：服务器端与客户端通讯必需标准化，服务器的变更并不会影响客户端。
• Uniform Interface（统一接口）：客户端与服务器端的通讯方法必需是统一的。
• Code on demand（按需执行代码？）：服务器端可以在上下文中执行代码或者脚本？

RESTful web service的样子

REST架构就是为了HTTP协议设计的。RESTful web services的核心概念是管理资源。资源是由URIs来表示，客户端使用HTTP当中的'POST, OPTIONS, GET, PUT, DELETE'等方法发送请求到服务器，改变相应的资源状态。

HTTP请求方法通常也十分合适去描述操作资源的动作：

HTTP方法	动作	例子
GET	获取资源信息	http://example.com/api/orders （检索订单清单）
GET	获取资源信息	http://example.com/api/orders/123 （检索订单 #123）
POST	创建一个次的资源	http://example.com/api/orders （使用带数据的请求，创建一个新的订单）
PUT	更新一个资源	http://example.com/api/orders/123 （使用带数据的请求，更新#123订单）
DELETE	删除一个资源	http://example.com/api/orders/123 删除订单#123

REST请求并不需要特定的数据格式，通常使用JSON作为请求体，或者URL的查询参数的一部份。

设计一个简单的web service

下面的任务将会练习设计以REST准则为指引，通过不同的请求方法操作资源，标识资源的例子。

我们将写一个To Do List 应用，并且设计一个web service。第一步，规划一个根URL，例如：

上面的URL包括了应用程序的名称、API版本，这是十分有用的，既提供了命名空间的划分，同时又与其它系统区分开来。版本号在升级新特性时十分有用，当一个新功能特性增加在新版本下面时，并不影响旧版本。

第二步，规划资源的URL，这个例子十分简单，只有任务清单。

规划如下：

HTTP方法	URI	动作
GET	http://[hostname]/todo/api/v1.0/tasks	检索任务清单
GET	http://[hostname]/todo/api/v1.0/tasks/[task_id]	检索一个任务
POST	http://[hostname]/todo/api/v1.0/tasks	创建一个新任务
PUT	http://[hostname]/todo/api/v1.0/tasks/[task_id]	更新一个已存在的任务
DELETE	http://[hostname]/todo/api/v1.0/tasks/[task_id]	删除一个任务

我们定义任务清单有以下字段：

• id:唯一标识。整型。
• title:简短的任务描述。字符串型。
• description:完整的任务描述。文本型。
• done:任务完成状态。布尔值型。

以上基本完成了设计部份，接下来我们将会实现它！

简单了解Flask框架

Flask好简单，但是又很强大的Python web 框架。这里有一系列教程Flask Mega-Tutorial series。（注：Django\Tornado\web.py感觉好多框:(）

在我们深入实现web service之前，让我们来简单地看一个Flask web 应用的结构示例。

这里都是在Unix-like（Linux,Mac OS X）操作系统下面的演示，但是其它系统也可以跑，例如windows下的Cygwin。可能命令有些不同吧。（注：忽略Windows吧。）

先使用virtualenv安装一个Flask的虚拟环境。如果没有安装virtualenv，开发python必备,最好去下载安装。https://pypi.python.org/pypi/virtualenv

$ mkdir todo-api

$ cd todo-api

$ virtualenv flask

New python executable in flask/bin/python

Installing setuptools............................done.

Installing pip...................done.

$ flask/bin/pip install flask

这样做好了一个Flask的开发环境，开始创建一个简单的web应用，在当前目录里面创建一个app.py文件：

#!flask/bin/python

from flask import Flask

app = Flask(__name__)

@app.route('/')

def index():

    return "Hello, World!"

if __name__ == '__main__':

    app.run(debug=True)

去执行app.py：

$ chmod a+x app.py

$ ./app.py

 * Running on http://127.0.0.1:5000/

 * Restarting with reloader

现在可以打开浏览器，输入http://localhost:5000去看看这个Hello,World!

好吧，十分简单吧。我们开始转换到RESTful service！

使用Python 和 Flask实现RESTful services

使用Flask建立web services超级简单。

当然，也有很多Flask extensions可以帮助建立RESTful services，但是这个例实在太简单了，不需要使用任何扩展。

这个web service提供增加，删除、修改任务清单，所以我们需要将任务清单存储起来。最简单的做法就是使用小型的数据库，但是数据库并不是本文涉及太多的。可以参考原文作者的完整教程。Flask Mega-Tutorial series

在这里例子我们将任务清单存储在内存中，这样只能运行在单进程和单线程中，这样是不适合作为生产服务器的，若非就必需使用数据库了。

现在我们准备实现第一个web service的入口点：

#!flask/bin/python

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/api/v1.0/tasks', methods=['GET'])

def get_tasks():

    return jsonify({'tasks': tasks})

if __name__ == '__main__':

    app.run(debug=True)

正如您所见，并没有改变太多代码。我们将任务清单存储在list内（内存），list存放两个非常简单的数组字典。每个实体就是我们上面定义的字段。

而 index 入口点有一个get_tasks函数与/todo/api/v1.0/tasks URI关联，只接受http的GET方法。

这个响应并非一般文本，是JSON格式的数据，是经过Flask框架的 jsonify模块格式化过的数据。

使用浏览器去测试web service并不是一个好的办法，因为要创建不同类弄的HTTP请求，事实上，我们将使用curl命令行。如果没有安装curl，快点去安装一个。

像刚才一样运行app.py。

打开一个终端运行以下命令：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 294

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 04:53:53 GMT

{

  "tasks": [

    {

      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",

      "done": false,

      "id": 1,

      "title": "Buy groceries"

    },

    {

      "description": "Need to find a good Python tutorial on the web",

      "done": false,

      "id": 2,

      "title": "Learn Python"

    }

  ]

}

这样就调用了一个RESTful service方法！

现在，我们写第二个版本的GET方法获取特定的任务。获取单个任务：

from flask import abort

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

第二个函数稍稍复杂了一些。任务的id包含在URL内，Flask将task_id参数传入了函数内。

通过参数，检索tasks数组。如果参数传过来的id不存在于数组内，我们需要返回错误代码404，按照HTTP的规定，404意味着是"Resource Not Found"，资源未找到。

如果找到任务在内存数组内，我们通过jsonify模块将字典打包成JSON格式，并发送响应到客户端上。就像处理一个实体字典一样。

试试使用curl调用：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks/2

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 151

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 05:21:50 GMT

{

  "task": {

    "description": "Need to find a good Python tutorial on the web",

    "done": false,

    "id": 2,

    "title": "Learn Python"

  }

}

$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3

HTTP/1.0 404 NOT FOUND

Content-Type: text/html

Content-Length: 238

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 05:21:52 GMT

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">

<title>404 Not Found</title>

<h1>Not Found</h1>

<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>

当我们请求#2 id的资源时，可以获取，但是当我们请求#3的资源时返回了404错误。并且返回了一段奇怪的HTML错误，而不是我们期望的JSON，这是因为Flask产生了默认的404响应。客户端需要收到的都是JSON的响应，因此我们需要改进404错误处理：

from flask import make_response

@app.errorhandler(404)

def not_found(error):

    return make_response(jsonify({'error': 'Not found'}), 404)

这样我们就得到了友好的API错误响应：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3

HTTP/1.0 404 NOT FOUND

Content-Type: application/json

Content-Length: 26

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 05:36:54 GMT

{

  "error": "Not found"

}

接下来我们实现 POST 方法，插入一个新的任务到数组中：

from flask import request

@app.route('/todo/api/v1.0/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

request.json里面包含请求数据，如果不是JSON或者里面没有包括title字段，将会返回400的错误代码。

当创建一个新的任务字典，使用最后一个任务id数值加1作为新的任务id（最简单的方法产生一个唯一字段）。这里允许不带description字段，默认将done字段值为False。

将新任务附加到tasks数组里面，并且返回客户端201状态码和刚刚添加的任务内容。HTTP定义了201状态码为“Created”。

测试上面的新功能：

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

HTTP/1.0 201 Created

Content-Type: application/json

Content-Length: 104

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 05:56:21 GMT

{

  "task": {

    "description": "",

    "done": false,

    "id": 3,

    "title": "Read a book"

  }

}

注意：如果使用原生版本的curl命令行提示符，上面的命令会正确执行。如果是在Windows下使用Cygwin bash版本的curl，需要将body部份添加双引号：

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

基本上在Windows中需要使用双引号包括body部份在内，而且需要三个双引号转义序列。

完成上面的事情，就可以看到更新之后的list数组内容：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 423

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 05:57:44 GMT

{

  "tasks": [

    {

      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",

      "done": false,

      "id": 1,

      "title": "Buy groceries"

    },

    {

      "description": "Need to find a good Python tutorial on the web",

      "done": false,

      "id": 2,

      "title": "Learn Python"

    },

    {

      "description": "",

      "done": false,

      "id": 3,

      "title": "Read a book"

    }

  ]

}

剩余的两个函数如下：

@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]})

@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函数需要检查所输入的参数，防止产生错误的bug。确保是预期的JSON格式写入数据库里面。

测试将任务#2的done字段变更为done状态：

$ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://localhost:5000/todo/api/v1.0/tasks/2

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 170

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 07:10:16 GMT

{

  "task": [

    {

      "description": "Need to find a good Python tutorial on the web",

      "done": true,

      "id": 2,

      "title": "Learn Python"

    }

  ]

}

改进Web Service接口

当前我们还有一个问题，客户端有可能需要从返回的JSON中重新构造URI，如果将来加入新的特性时，可能需要修改客户端。（例如新增版本。）

我们可以返回整个URI的路径给客户端，而不是任务的id。为了这个功能，创建一个小函数生成一个“public”版本的任务URI返回：

from flask import url_for

def make_public_task(task):

    new_task = {}

    for field in task:

        if field == 'id':

            new_task['uri'] = url_for('get_task', task_id=task['id'], _external=True)

        else:

            new_task[field] = task[field]

    return new_task

通过Flask的url_for模块，获取任务时，将任务中的id字段替换成uri字段，并且把值改为uri值。

当我们返回包含任务的list时，通过这个函数处理后，返回完整的uri给客户端：

@app.route('/todo/api/v1.0/tasks', methods=['GET'])

def get_tasks():

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

现在看到的检索结果：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 406

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 18:16:28 GMT

{

  "tasks": [

    {

      "title": "Buy groceries",

      "done": false,

      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",

      "uri": "http://localhost:5000/todo/api/v1.0/tasks/1"

    },

    {

      "title": "Learn Python",

      "done": false,

      "description": "Need to find a good Python tutorial on the web",

      "uri": "http://localhost:5000/todo/api/v1.0/tasks/2"

    }

  ]

}

这种办法避免了与其它功能的兼容，拿到的是完整uri而不是一个id。

RESTful web service的安全认证

我们已经完成了整个功能，但是我们还有一个问题。web service任何人都可以访问的，这不是一个好主意。

当前service是所有客户端都可以连接的，如果有别人知道了这个API就可以写个客户端随意修改数据了。 大多数教程没有与安全相关的内容，这是个十分严重的问题。

最简单的办法是在web service中，只允许用户名和密码验证通过的客户端连接。在一个常规的web应用中，应该有登录表单提交去认证，同时服务器会创建一个会话过程去进行通讯。这个会话过程id会被存储在客户端的cookie里面。不过这样就违返了我们REST中无状态的规则，因此，我们需求客户端每次都将他们的认证信息发送到服务器。

为此我们有两种方法表单认证方法去做，分别是 Basic 和 Digest。

这里有有个小Flask extension可以轻松做到。首先需要安装 Flask-HTTPAuth ：

$ flask/bin/pip install flask-httpauth

假设web service只有用户 ok 和密码为 python 的用户接入。下面就设置了一个Basic HTTP认证：

from flask.ext.httpauth import HTTPBasicAuth

auth = HTTPBasicAuth()

@auth.get_password

def get_password(username):

    if username == 'ok':

        return 'python'

    return None

@auth.error_handler

def unauthorized():

    return make_response(jsonify({'error': 'Unauthorized access'}), 401)

get_password函数是一个回调函数，获取一个已知用户的密码。在复杂的系统中，函数是需要到数据库中检查的，但是这里只是一个小示例。

当发生认证错误之后，error_handler回调函数会发送错误的代码给客户端。这里我们自定义一个错误代码401，返回JSON数据，而不是HTML。

将@auth.login_required装饰器添加到需要验证的函数上面：

@app.route('/todo/api/v1.0/tasks', methods=['GET'])

@auth.login_required

def get_tasks():

    return jsonify({'tasks': tasks})

现在，试试使用curl调用这个函数：

$ curl -i http://localhost:5000/todo/api/v1.0/tasks

HTTP/1.0 401 UNAUTHORIZED

Content-Type: application/json

Content-Length: 36

WWW-Authenticate: Basic realm="Authentication Required"

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 06:41:14 GMT

{

  "error": "Unauthorized access"

}

这里表示了没通过验证，下面是带用户名与密码的验证：

$ curl -u ok:python -i http://localhost:5000/todo/api/v1.0/tasks

HTTP/1.0 200 OK

Content-Type: application/json

Content-Length: 316

Server: Werkzeug/0.8.3 Python/2.7.3

Date: Mon, 20 May 2013 06:46:45 GMT

{

  "tasks": [

    {

      "title": "Buy groceries",

      "done": false,

      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",

      "uri": "http://localhost:5000/todo/api/v1.0/tasks/1"

    },

    {

      "title": "Learn Python",

      "done": false,

      "description": "Need to find a good Python tutorial on the web",

      "uri": "http://localhost:5000/todo/api/v1.0/tasks/2"

    }

  ]

}

这个认证extension十分灵活，可以随指定需要验证的APIs。

为了确保登录信息的安全，最好的办法还是使用https加密的通讯方式，客户端与服务器端传输认证信息都是加密过的，防止第三方的人去看到。

当使用浏览器去访问这个接口，会弹出一个丑丑的登录对话框，如果密码错误就回返回401的错误代码。为了防止浏览器弹出验证对话框，客户端应该处理好这个登录请求。

有一个小技巧可以避免这个问题，就是修改返回的错误代码401。例如修改成403（”Forbidden“）就不会弹出验证对话框了。

@auth.error_handler

def unauthorized():

    return make_response(jsonify({'error': 'Unauthorized access'}), 403)

当然，同时也需要客户端知道这个403错误的意义。

最后

还有很多办法去改进这个web service。

事实上，一个真正的web service应该使用真正的数据库。使用内存数据结构有非常多的限制，不要用在实际应用上面。

另外一方面，处理多用户。如果系统支持多用户认证，则任务清单也是对应多用户的。同时我们需要有第二种资源，用户资源。当用户注册时使用POST请求。使用GET返回用户信息到客户端。使用PUT请求更新用户资料，或者邮件地址。使用DELETE删除用户账号等。

通过GET请求检索任务清单时，有很多办法可以进扩展。第一，可以添加分页参数，使客户端只请求一部份数据。第二，可以添加筛选关键字等。所有这些元素可以添加到URL上面的参数。