| 编辑推荐: |
| 本文来自于blog.miguelgrinberg.com，文章从安装部署到设计到对web服务增加安全机制娓娓道来。 |
|
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进行测试。
|
