Flask-RESTful 快速入門

hello world

from flask import Flask
from flask_restful import Resource, Api

app = Flask(__name__)
api = Api(app)

class HelloWorld(Resource):
    def get(self):
        return {'hello': 'world'}

api.add_resource(HelloWorld, '/')

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

資源

資源 (Resource) 是 Flask-RESTful 中最主要的概念. 資源就是 Flask pluggable views 的子類, 因此可以很容易的定義 HTTP 方法. 一個簡單的 CRUD 程序看起來如下:

from flask import Flask, request
from flask_restful import Resource, Api

app = Flask(__name__)
api = Api(app)

todos = {}

class TodoSimple(Resource):
    def get(self, todo_id):
        return {todo_id: todos[todo_id]}

    def put(self, todo_id):
        todos[todo_id] = request.form['data']
        return {todo_id: todos[todo_id]}

// 將　TodoSimple 掛載在 '/<string:todo_id>' 路由上
api.add_resource(TodoSimple, '/<string:todo_id>')

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

Flask-RESTful 能夠處理各種類型返回值, 將其轉換成合適的響應.

class Todo1(Resource):
    def get(self):
        # 默認 HTTP Status Code 為 200
        return {'task': 'Hello world'}

class Todo2(Resource):
    def get(self):
        # 設置 HTTP Status Code 為 201
        return {'task': 'Hello world'}, 201

class Todo3(Resource):
    def get(self):
        # 設置 HTTP Status Code 為 201 並返回自定義 headers
        return {'task': 'Hello world'}, 201, {'Etag': 'some-opaque-string'}

Endpoints

一個資源可以掛載在多個路由上:

api.add_resource(HelloWorld,
    '/',
    '/hello')

可以將路由作為參數

api.add_resource(Todo,
    '/todo/<int:todo_id>', endpoint='todo_ep')

參數解析

雖然 Flask 提供了簡單的方法訪問請求數據, 但是對於表單數據處理還是很痛苦. Flask-RESTful 提供了 reqparse 一種類似 argparse 參數解析機制以簡化參數處理.

from flask_restful import reqparse

parser = reqparse.RequestParser()
parser.add_argument('rate', type=int, help='Rate to charge for this resource')
// strict 如果提供未定義的參數, 那么就拋出異常
args = parser.parse_args(strict=True)

數據格式化

Flask 能很好的處理 Python 內置的數據結構, Flask-RESTful 提供了處理自定義對象的 fields 模塊和 marshal_with() 方法

from collections import OrderedDict
from flask_restful import fields, marshal_with

resource_fields = {
    'task':   fields.String,
    'uri':    fields.Url('todo_ep')
}

class TodoDao(object):
    def __init__(self, todo_id, task):
        self.todo_id = todo_id
        self.task = task

        # This field will not be sent in the response
        self.status = 'active'

class Todo(Resource):
    @marshal_with(resource_fields)
    def get(self, **kwargs):
        return TodoDao(todo_id='my_todo', task='Remember the milk')

上例中 get 返回一個 TodoDao 對象, 使用 marshal_with 裝飾器根據 resource_fields 序列化這個對象. 這里只有一個字段被序列化, 就是 task.

總結

Flask-RESTful 的使用基本上就是用 reqparse 處理請求, 使用 fields 來處理返回值, 輸出響應.

完整的例子

from flask import Flask
from flask_restful import reqparse, abort, Api, Resource

app = Flask(__name__)
api = Api(app)

TODOS = {
    'todo1': {'task': 'build an API'},
    'todo2': {'task': '?????'},
    'todo3': {'task': 'profit!'},
}


def abort_if_todo_doesnt_exist(todo_id):
    if todo_id not in TODOS:
        abort(404, message="Todo {} doesn't exist".format(todo_id))

parser = reqparse.RequestParser()
parser.add_argument('task')


# Todo
# shows a single todo item and lets you delete a todo item
class Todo(Resource):
    def get(self, todo_id):
        abort_if_todo_doesnt_exist(todo_id)
        return TODOS[todo_id]

    def delete(self, todo_id):
        abort_if_todo_doesnt_exist(todo_id)
        del TODOS[todo_id]
        return '', 204

    def put(self, todo_id):
        args = parser.parse_args()
        task = {'task': args['task']}
        TODOS[todo_id] = task
        return task, 201


# TodoList
# shows a list of all todos, and lets you POST to add new tasks
class TodoList(Resource):
    def get(self):
        return TODOS

    def post(self):
        args = parser.parse_args()
        todo_id = int(max(TODOS.keys()).lstrip('todo')) + 1
        todo_id = 'todo%i' % todo_id
        TODOS[todo_id] = {'task': args['task']}
        return TODOS[todo_id], 201

##
## Actually setup the Api resource routing here
##
api.add_resource(TodoList, '/todos')
api.add_resource(Todo, '/todos/<todo_id>')


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