flask-restx基于flask的restful风格的插件

前言

兄弟们在做web开发服务的时候我推荐大家使用flask-restx这个插件，对比falsk而言，flask只适合做比较小的微服务组件，而不是一个完成的web应用，且flask-restx集成swagger和restful更方便的管理我们的api服务，使用方法和flask并无太大的差异，下面我们来具体了解flask-restx怎么使用的吧。

环境安装

Flask==2.0.1
flask-restx==0.5.1
Werkzeug==2.0.1

使用

• 创建和运行实例

from flask import Flask
from flask_restx import Api, Resource

api = Api()

app = Flask(__name__)
api.init_app(app)

@api.route('/hello',strict_slashes=False)
class HelloWorld(Resource):
    def get(self):
 	    # 如果使用模板的块，需要使用 make_response
        # return make_response(render_template('index.html', data=res), 200)
        
        # 使用 jsonify 是为了返回json数据的同时，相比于 json.dumps() 其会自动修改 content-type 为 application/json
        # 另外，如果使用 jsonify()的同时，还想自定义返回状态码，可以使用 make_response(jsonify(data=data), 201)
        return jsonify({'hello': 'world'})
    
    def post(self):
        pass
    
    def put(self):
        pass
    
    def delete(self):
        pass

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

• 资源路径限定

@api.route('/<string:todo_id>')
@api.route('/<int:todo_id>')

• 响应（数据，状态码，响应头）

@api.route('/todo')
class Todo(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(HelloWorld, '/', '/hello')
api.add_resource(Todo, '/todo/<int:todo_id>', endpoint='todo_ep')

• 参数解析

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

args = parser.parse_args(strict=True) 确保参数没有解析前返回错误

$ curl -d 'rate2=foo' http://127.0.0.1:5000/todos
{
    "message": "Unknown arguments: rate2"
}

• 数据格式化

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)

指定返回的格式，提取唯一的task字段，uri它接受端点（endpoint）名称作为参数并且在响应中为该端点生成一个 URL。许多你需要的字段类型都已经包含在内。请参阅 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)

参考

• api.model使用
• restful实践