Pythonflask-restful框架讲解

flask-restful 安装

pip install flask-restful

flask-restful使⽤

简单上⼿

from flask import Flask

from flask_restful import Resource, Apiapp = 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 http://127.0.0.1:5000/{\"hello\": \"world\

Resourceful 路由

Flask-RESTful 提供的主要构建块是资源。资源构建在 Flask 可插⼊视图之上，只需在资源上定义⽅法，就可以轻松访问多个 HTTP ⽅法。⼀个 todo 应⽤程序的基本 CRUD 资源是这样的:

from flask import Flask, request

from flask_restful import Resource, Apiapp = 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, '/')

if name == 'main':app.run(debug=True)

运⾏结果：

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://localhost:5000/todo1 -d \"data=Remember the milk\" -X PUT{

\"todo1\": \"Remember the milk\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://localhost:5000/todo1{

\"todo1\": \"Remember the milk\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://localhost:5000/todo2 -d \"data=Change my brakepads\" -X PUT{

\"todo2\": \"Change my brakepads\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://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, Apiapp = Flask(name)api = Api(app)

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

class Todo2(Resource):def get(self):

# Set the response code to 201return {'task': 'Hello world'}, 201

class Todo3(Resource):def get(self):

# Set the response code to 201 and return custom headersreturn {'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 http://127.0.0.1:5000/todo1 HTTP/1.0 200 OK

Content-Type: application/jsonContent-Length: 30

Server: Werkzeug/1.0.1 Python/3.9.2Date: Fri, 05 Mar 2021 16:08:28 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/', endpoint='todo_ep')

演⽰代码：

from flask import Flask

from flask_restful import Resource, Apiapp = 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 OKreturn {'task': 'Hello world'}

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

api.add_resource(Todo, '/todo/', endpoint='todo_ep')

if name == 'main':app.run(debug=True)

演⽰结果：

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://127.0.0.1:5000/{

\"hello\": \"world\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://127.0.0.1:5000/hello{

\"hello\": \"world\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://127.0.0.1:5000/todo/1{

\"task\": \"Hello world\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl http://127.0.0.1:5000/todo/2{

\"task\": \"Hello world\"}

参数解析

虽然 Flask 可以⽅便地访问请求数据(即 querystring 或 POST 表单编码的数据) ，但验证表单数据仍然是⼀件痛苦的事情。使⽤类似于 argparse 的库对请求数据验证提供内置⽀持。

from flask import Flask

from flask_restful import reqparse, Api, Resourceapp = 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 OKreturn {'task': 'Hello world'}

api.add_resource(Todo,'/todos' )

if name == 'main':app.run(debug=True)

chenshifengdeMacBook-Pro:~ chenshifeng$ curl -d 'rate=100' http://127.0.0.1:5000/todos{

\"task\": \"Hello world\"}

chenshifengdeMacBook-Pro:~ chenshifeng$ curl -d 'rate=foo' http://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' http://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, Apiapp = 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_idself.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 http://127.0.0.1:5000/todo{

\"task\": \"Remember the milk\

\"uri\": \"/todo\"}

完整例⼦

from flask import Flask

from flask_restful import reqparse, abort, Api, Resourceapp = 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 itemclass 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 tasksclass 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/')

if name == 'main':app.run(debug=True)

获取列表

$ curl http://localhost:5000/todos{

\"todo1\": {

\"task\": \"build an API\" },

\"todo2\": {

\"task\": \"\" },

\"todo3\": {

\"task\": \"profit!\" }}

获取⼀个单独的任务

$ curl http://localhost:5000/todos/todo3{

\"task\": \"profit!\"}

删除⼀个任务

$ curl http://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 http://localhost:5000/todos -d \"task=something new\" -X POST -vNote: 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 http://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 http://localhost:5000/todos{

\"todo1\": {

\"task\": \"build an API\" },

\"todo3\": {

\"task\": \"something different\" },

\"todo4\": {

\"task\": \"something new\" }}