Python flask-restful框架講解

Flask-RESTful 是一個 Flask 擴展,它添加了快速構建 REST APIs 的支持。它當然也是一個能夠跟你現有的ORM/庫協同工作的輕量級的擴展。Flask-RESTful 鼓勵以最小設置的最佳實踐。如果你熟悉 Flask 的話,Flask-RESTful 應該很容易上手。
關於flask的使用,參考我的之前的博客://blog.csdn.net/shifengboy/article/details/114274271

flask-restful官方文檔://flask-restful.readthedocs.io/en/lates
中文文檔://www.pythondoc.com/Flask-RESTful/

flask-restful 安裝

pip install flask-restful

flask-restful使用

簡單上手

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)

運行結果:

$ curl //127.0.0.1:5000/
{"hello": "world"}

Resourceful 路由

Flask-RESTful 提供的主要構建塊是資源。資源構建在 Flask 可插入視圖之上,只需在資源上定義方法,就可以輕鬆訪問多個 HTTP 方法。一個 todo 應用程序的基本 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]}

api.add_resource(TodoSimple, '/<string:todo_id>')

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

運行結果:

chenshifengdeMacBook-Pro:~ chenshifeng$ curl //localhost:5000/todo1 -d "data=Remember the milk" -X PUT
{
    "todo1": "Remember the milk"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl //localhost:5000/todo1
{
    "todo1": "Remember the milk"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl //localhost:5000/todo2 -d "data=Change my brakepads" -X PUT
{
    "todo2": "Change my brakepads"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl //localhost:5000/todo2
{
    "todo2": "Change my brakepads"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ 

Restful 能夠從 view 方法中理解多種返回值。類似於 Flask,你可以返回任何可迭代的並且它將被轉換成一個響應,包括原始 Flask 響應對象。還支持使用多個返回值設置響應代碼和響應頭,如下所示:

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:chenshifeng
@file:flask_restful_demo.py
@time:2021/03/05
"""
from flask import Flask, request
from flask_restful import Resource, Api

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

class Todo1(Resource):
    def get(self):
        # Default to 200 OK
        return {'task': 'Hello world'}

class Todo2(Resource):
    def get(self):
        # Set the response code to 201
        return {'task': 'Hello world'}, 201

class Todo3(Resource):
    def get(self):
        # Set the response code to 201 and return custom headers
        return {'task': 'Hello world'}, 201, {'Etag': 'some-opaque-string'}

api.add_resource(Todo1,'/todo1')
api.add_resource(Todo2,'/todo2')
api.add_resource(Todo3,'/todo3')

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

運行結果:

chenshifengdeMacBook-Pro:~ chenshifeng$ curl -i  //127.0.0.1:5000/todo1 
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 30
Server: Werkzeug/1.0.1 Python/3.9.2
Date: Fri, 05 Mar 2021 16:08:28 GMT

{
    "task": "Hello world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl -i  //127.0.0.1:5000/todo2
HTTP/1.0 201 CREATED
Content-Type: application/json
Content-Length: 30
Server: Werkzeug/1.0.1 Python/3.9.2
Date: Fri, 05 Mar 2021 16:08:32 GMT

{
    "task": "Hello world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl -i  //127.0.0.1:5000/todo3
HTTP/1.0 201 CREATED
Content-Type: application/json
Content-Length: 30
Etag: some-opaque-string
Server: Werkzeug/1.0.1 Python/3.9.2
Date: Fri, 05 Mar 2021 16:08:34 GMT

{
    "task": "Hello world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ 

Endpoints 端點

很多時候,在一個 API 中,你的資源會有多個 url。可以將多個 url 傳遞給 Api 對象上的 add _ resource ()方法。每一個都將被路由到Resource

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

您還可以將路徑的某些部分作為變量匹配到Resource。

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

演示代碼:

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


class Todo(Resource):
    def get(self, todo_id):
        # Default to 200 OK
        return {'task': 'Hello world'}


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

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

演示結果:

chenshifengdeMacBook-Pro:~ chenshifeng$ curl  //127.0.0.1:5000/
{
    "hello": "world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl  //127.0.0.1:5000/hello
{
    "hello": "world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl  //127.0.0.1:5000/todo/1
{
    "task": "Hello world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl  //127.0.0.1:5000/todo/2
{
    "task": "Hello world"
}

參數解析

雖然 Flask 可以方便地訪問請求數據(即 querystring 或 POST 表單編碼的數據) ,但驗證表單數據仍然是一件痛苦的事情。使用類似於 argparse 的庫對請求數據驗證提供內置支持。

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

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

parser = reqparse.RequestParser()
parser.add_argument('rate', type=int, help='Rate to charge for this resource')

class Todo(Resource):
    def post(self):
        args = parser.parse_args()
        print(args)
        # Default to 200 OK
        return {'task': 'Hello world'}


api.add_resource(Todo,'/todos' )

if __name__ == '__main__':
    app.run(debug=True)
chenshifengdeMacBook-Pro:~ chenshifeng$ curl -d 'rate=100' //127.0.0.1:5000/todos
{
    "task": "Hello world"
}
chenshifengdeMacBook-Pro:~ chenshifeng$ curl -d 'rate=foo' //127.0.0.1:5000/todos
{
    "message": {
        "rate": "Rate to charge for this resource"
    }
}

與 argparse 模塊不同,reqparse. RequestParser.parse _ args ()返回 Python 字典,而不是自定義數據結構。

輸入模塊提供了許多常用的轉換函數,例如 inputs.date ()和 inputs.url ()。
使用 strict = True 調用 parse _ args 可以確保在請求包含您的解析器沒有定義的參數時拋出錯誤。

args = parser.parse_args(strict=True)
$ curl -d 'rate2=foo' //127.0.0.1:5000/todos
{
    "message": "Unknown arguments: rate2"
}

數據格式化

默認情況下,在你的返回迭代中所有字段將會原樣呈現。儘管當你剛剛處理 Python 數據結構的時候,覺得這是一個偉大的工作,但是當實際處理它們的時候,會覺得十分沮喪和枯燥。為了解決這個問題,Flask-RESTful 提供了 fields 模塊和 marshal_with() 裝飾器。類似 Django ORM 和 WTForm,你可以使用 fields 模塊來在你的響應中格式化結構。

from flask import Flask
from flask_restful import fields, marshal_with, Resource, Api

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

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

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

api.add_resource(Todo,'/todo')

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

上面的例子接受一個 python 對象並準備將其序列化。marshal_with() 裝飾器將會應用到由 resource_fields 描述的轉換。從對象中提取的唯一字段是 task。fields.Url 域是一個特殊的域,它接受端點(endpoint)名稱作為參數並且在響應中為該端點生成一個 URL。許多你需要的字段類型都已經包含在內。請參閱 fields 指南獲取一個完整的列表。

$ curl  //127.0.0.1:5000/todo
{
    "task": "Remember the milk",
    "uri": "/todo"
}

完整例子

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)

獲取列表

$ curl //localhost:5000/todos
{
    "todo1": {
        "task": "build an API"
    },
    "todo2": {
        "task": "?????"
    },
    "todo3": {
        "task": "profit!"
    }
}

獲取一個單獨的任務

$ curl //localhost:5000/todos/todo3
{
    "task": "profit!"
}

刪除一個任務

$ curl //localhost:5000/todos/todo2 -X DELETE -v
*   Trying ::1...
* TCP_NODELAY set
* Connection failed
* connect to ::1 port 5000 failed: Connection refused
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 5000 (#0)
> DELETE /todos/todo2 HTTP/1.1
> Host: localhost:5000
> User-Agent: curl/7.64.1
> Accept: */*
> 
* HTTP 1.0, assume close after body
< HTTP/1.0 204 NO CONTENT
< Content-Type: application/json
< Server: Werkzeug/1.0.1 Python/3.9.2
< Date: Sat, 06 Mar 2021 03:29:33 GMT
< 
* Closing connection 0

增加一個新的任務

$ curl //localhost:5000/todos -d "task=something new" -X POST -v
Note: Unnecessary use of -X or --request, POST is already inferred.
*   Trying ::1...
* TCP_NODELAY set
* Connection failed
* connect to ::1 port 5000 failed: Connection refused
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 5000 (#0)
> POST /todos HTTP/1.1
> Host: localhost:5000
> User-Agent: curl/7.64.1
> Accept: */*
> Content-Length: 18
> Content-Type: application/x-www-form-urlencoded
> 
* upload completely sent off: 18 out of 18 bytes
* HTTP 1.0, assume close after body
< HTTP/1.0 201 CREATED
< Content-Type: application/json
< Content-Length: 32
< Server: Werkzeug/1.0.1 Python/3.9.2
< Date: Sat, 06 Mar 2021 03:31:02 GMT
< 
{
    "task": "something new"
}
* Closing connection 0

更新一個任務

$  curl //localhost:5000/todos/todo3 -d "task=something different" -X PUT -v
*   Trying ::1...
* TCP_NODELAY set
* Connection failed
* connect to ::1 port 5000 failed: Connection refused
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 5000 (#0)
> PUT /todos/todo3 HTTP/1.1
> Host: localhost:5000
> User-Agent: curl/7.64.1
> Accept: */*
> Content-Length: 24
> Content-Type: application/x-www-form-urlencoded
> 
* upload completely sent off: 24 out of 24 bytes
* HTTP 1.0, assume close after body
< HTTP/1.0 201 CREATED
< Content-Type: application/json
< Content-Length: 38
< Server: Werkzeug/1.0.1 Python/3.9.2
< Date: Sat, 06 Mar 2021 03:32:44 GMT
< 
{
    "task": "something different"
}
* Closing connection 0

獲取最新列表

$ curl //localhost:5000/todos
{
    "todo1": {
        "task": "build an API"
    },
    "todo3": {
        "task": "something different"
    },
    "todo4": {
        "task": "something new"
    }
}