Flask 初识
一个最小的API以下定义了一个简单的get和post请求get请求,返回Hello World!post请求,发送任务信息的json数据,如:{“id”:“1”,“info”:“firstTask”},服务根据请求传输的json数据生成任务并添加到任务列表import jsonfrom flask import Flask, abort, request, jsonifyapp = Flask(_
一个最小的API
以下定义了一个简单的get和post请求
get请求,返回Hello World!
post请求,发送任务信息的json数据,如:{“id”:“1”,“info”:“firstTask”},服务根据请求传输的json数据生成任务并添加到任务列表
import json
from flask import Flask, abort, request, jsonify
app = Flask(__name__)#创建实例
tasks = []
#无参接口
@app.route('/HelloWorld')
def hello_world():
return "Hello World!"
@app.route('/addTask/', methods=['POST'])
def add_task():
if not request.json or 'id' not in request.json or 'info' not in request.json:
abort(400)
task = {'id': int(request.json['id']),'info': request.json['info']}#获取post请求参数,request.json[]
tasks.append(task)#添加任务到任务列表
return jsonify({'result': 'success'})
if __name__ == '__main__':
app.run(debug=True) # 可以设置ip和端口,默认5000端口
flask-restful
Flask-RESTful 提供的最主要的基础就是资源(resources)。资源(Resources)是构建在 Flask 可拔插视图 之上,只要在你的资源(resource)上定义方法就能够容易地访问多个 HTTP 方法。主要体现在可以设置一个url的不同请求方式对应的操作,而无需自定义多个接口路由信息。
from flask import Flask, request
from flask_restful import Api, Resource
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)
上述代码TodoSimple就是一个资源,在这个资源下,我们可以定义多个方法,上述代码定义了get put两个方法,get请求根据路径中的"todo_id"获取todos字典中相应值,put请求补充todos中相应值并返回。
测试
python自带的requests库
from requests import put, get
put('http://localhost:5000/todo1', data={'data': 'test data'}).json()
# output
{'id1': 'test data'}
通过curl
curl http://localhost:5000/todo1 -d "data=test data" -X PUT # put
curl http://127.0.0.1:5000/todo1 -I # get
endpoint 端点
很多时候在一个API中,你的资源可以通过多个URL访问。你可以把多个URL传给Api对象的add_resource()方法。每一个URL可以访问到你的资源。这里的对应关系为,URL对应到端点,端点对应到具体的视图函数。
Flask中endpoint的理解
api.add_resource(HelloWorld,
'/',
'/hello')
参数解析
Flask-RESTful内置了支持验证请求数据,它使用了一个类似argparse的库。(返回字典)
from flask_restful import reqparse
parser = reqparse.RequestParser()
parser.add_argument('rate', type=int, help='Rate to charge for this resource')
args = parser.parse_args()
如果一个参数没有通过校验,则Flask-RESTful将以一个400的错误请求以及高亮的错误信息响应。
输入模块提供许多常用的转换函数,像inputs.date()和inputs.url()。
调用parse_args强制strict = True能够确保当请求包含了你的解析器中未定义的参数时引发一个异常。
app = Flask(__name__)
api = Api(app)
parser = reqparse.RequestParser()
parser.add_argument('num', type=int, help='number')
todos = {}
class TodoSimple(Resource):
def get(self, todo_id):
return {todo_id: todos[todo_id]}
def put(self, todo_id):
args = parser.parse_args(strict=True)
todos[todo_id] = request.form['num']
return {todo_id: todos[todo_id]}
api.add_resource(TodoSimple, '/<string:todo_id>')
if __name__ == '__main__':
app.run(debug=True)
127.0.0.1 - - [12/Jun/2020 14:32:07] "PUT /id1 HTTP/1.1" 400 -
使用gevent
Flask 是单线程运行,如果在某个页面中执行了一些耗时的工作,那么程序就会在这里等待,无法响应其他的请求
考虑使用 gevent 非阻塞的运行服务器程序。在引入 gevent 前,可以在程序最开始执行的位置引入猴子补丁 gevent.monkey,这能修改 python 默认的 IO 行为,让标准库变成协作式(cooperative)的 API。注意引入 gevent 后,不能再用原来的方式启动我们的 web 应用了
# app.py
from gevent import monkey
monkey.patch_all() # 打上猴子补丁
from flask import flask
...
if __name__ == '__main__':
from gevent import pywsgi
app.debug = True
server = pywsgi.WSGIServer( ('127.0.0.1', 5000 ), app )
server.serve_forever()
这块不详细描述,有需要
在 Flask 应用中使用 gevent
WARNING: This is a development server. Do not use it in a production deployment. falsk WSGI
异步执行任务
在开发中可能遇到一些执行时间长的任务,需要先返回请求的响应状态,长时间任务在后台进行。可以通过进程池来实现。
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(1)
task1 = executor.submit(func, ()) # 执行该语句后,请求可以返回响应
线程池的使用参考 [python] ThreadPoolExecutor线程池
Error
调试程序经常会遇到:OSError: [Errno 98] Address already in use
,这是因为上次程序的进程没有停止
通过 netstat -tunlp
获取被占用的地址和端口的进程id pid
通过kill -9 pid
,停掉该进程
建议停止程序通过Ctrl + C
而不是Ctrl + Z
OSError: [Errno 98] Address already in use解决办法
未完待续…
主要参考
Flask扩展系列之Flask-RESTful
Python中Flask-RESTful编写API接口—小白入门详解
更多推荐
所有评论(0)