输入模块提供了许多常用的转换函数,例如 inputs.date ()和 inputs.url ()。
使用 strict = True 调用 parse _ args 可以确保在请求包含您的解析器没有定义的参数时抛出错误。
默认情况下,在你的返回迭代中所有字段将会原样呈现。尽管当你刚刚处理 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 :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 :5000/todos { "todo1": { "task": "build an API" }, "todo2": { "task": "?????" }, "todo3": { "task": "profit!" } }获取一个单独的任务
$ curl :5000/todos/todo3 { "task": "profit!" }删除一个任务
$ curl :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 :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