“所有的道别中，我最喜欢的还是明天见。”
前段时间写了有关蓝图的部分，今天加入新的知识restplus。这一部分的基础知识自行百度就好了，后面有机会我会上传一份资源，是对这部分知识的总结和一些我的思考。话不多扯，我们从代码出发，来讲讲这一部分的知识是怎么用的。
restplus的理论和基础部分，大家可以自行百度或者下载我已经上传的资源——python学习模块.rar.
对了这个地方有一个小细节，我们一个文件下面可能会有很多个项目文件，在项目文件运行时记得把文件设置成源目录（就是灰色变成了蓝色），方法很简单：
app:
#创建app from werkzeug.middleware.proxy_fix import ProxyFix from flask import Flask, request from methods_move import methods_face def create_app(): app = Flask(__name__) # handle proxy server headers # 处理代理服务器头 app.wsgi_app = ProxyFix(app.wsgi_app) app.register_blueprint(methods_face.api_v1) return applogger_error:
import time import logging def logger_error(): # 创建一个logger global logger logger = logging.getLogger() logger.setLevel(logging.INFO) # Log等级总开关 # 创建一个handler，用于写入日志文件 rq = time.strftime('%Y%m%d%H%M', time.localtime(time.time())) log_path = 'C:/Users/BBD' + '/Logs/' log_name = log_path + rq + '.log' logfile = log_name fh = logging.FileHandler(logfile, mode='w') fh.setLevel(logging.DEBUG) # 输出到file的log等级的开关 # 定义handler的输出格式 formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s") fh.setFormatter(formatter) logger.addHandler(fh)database:
import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection def mysql_pool(): POOL = PooledDB( creator=pymysql, # 使用链接数据库的模块 maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数 mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建 maxcached=5, # 链接池中最多闲置的链接，0和None不限制 maxshared=3, # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。 blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错 maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制 setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always host='127.0.0.1', port=3306, user='root', password='mysql123456789', database='BASEONE', charset='utf8' ) return POOLmethods_face:
from flask import Flask, Blueprint from flask_restplus import Api, Resource, fields from database import mysql_pool import logger_error from flask_restplus import reqparse api_v1 = Blueprint('api', __name__) api = Api(api_v1, version='1.0', title='TodoMVC API', description='A simple TodoMVC API', ) ns = api.namespace('todos', description='TODO operations') todo = api.model('Todo', { 'id': fields.Integer(readOnly=True, description='The task unique identifier'),#整型字段 'author': fields.String(required=True),#String字段 'age': fields.Integer(required=True), 'current_address': fields.String(required=True), 'times_awards': fields.Integer(required=True) }) parser = reqparse.RequestParser(bundle_errors=True) parser.add_argument('id', type=int, required=True) parser.add_argument('author', type=str, required=True) parser.add_argument('age', type=int, required=True) parser.add_argument('current_address', type=str, required=True) parser.add_argument('times_awards', type=int, required=True) POOL = mysql_pool() @ns.route('/create_todo') class Todo(Resource): @ns.doc('create_todo') @ns.expect(todo) @ns.marshal_with(todo, code=201) def post(self): '''创建一个新的task''' # return SQL.create(api.payload), 201 # data = json.loads(api.payload) todo = {} id = todo['id'] = api.payload['id'] author = todo['author'] = api.payload['author'] age = todo['age'] = api.payload['age'] current_address = todo['current_address'] = api.payload['current_address'] times_awards = todo['times_awards'] = api.payload['times_awards'] conn = POOL.connection() cursor = conn.cursor() sql_insert = "INSERT INTO author (id,author,age,current_address,times_awards) VALUES (%s,%s,%s,%s,%s)" cursor.execute(sql_insert, (id, author, age, current_address, times_awards)) conn.commit() conn.close() return todo @ns.route('/get_todo/<int:id>') class TodoSimple(Resource): @api.doc('get_todo') @ns.marshal_with(todo) def get(self, id): '''获取id指定的todo项''' try: conn = POOL.connection() cursor = conn.cursor() sql_select = "select * from author where id = %s" if cursor.execute(sql_select, id): info = cursor.fetchall() conn.close() jsondata = [] for row in info: result = {} result["id"] = row[0] result["author"] = row[1] result["age"] = row[2] result["current_address"] = row[3] result["times_awards"] = row[4] jsondata.append(result) return jsondata else: api.abort(404, "Todo {} doesn't exist".format(id)) return todo except: logger_error() pass @ns.route('/delete_todo/<int:id>') class TodoSimple(Resource): @ns.doc('delete_todo') @ns.response(204, 'Todo deleted') def delete(self, id): '''根据id删除对应的task''' try: conn = POOL.connection() cursor = conn.cursor() sql_drop = "DELETE FROM author WHERE id = %s" cursor.execute(sql_drop, id) conn.commit() conn.close() return '', 204 except: logger_error() return '', 404 @ns.route('/update_todo/<int:id>') class TodoSimple(Resource): @ns.doc('update_todo') @ns.expect(todo) @ns.marshal_with(todo) def put(self, id): '''更新id指定的task''' try: todo = {} if id == api.payload['id']: todo['id'] = id author = todo['author'] = api.payload['author'] age = todo['age'] = api.payload['age'] current_address = todo['current_address'] = api.payload['current_address'] times_awards = todo['times_awards'] = api.payload['times_awards'] conn = POOL.connection() cursor = conn.cursor() sql_update = "UPDATE author SET author=%s,age=%s,current_address=%s,times_awards=%s WHERE id = %s" cursor.execute(sql_update, (author, age, current_address, times_awards, id)) conn.commit() conn.close() return todo except: pass logger_error() return '', 404运行程序——manager:
from app import create_app flask_app = create_app() if __name__ == '__main__': flask_app.run(host='0.0.0.0', debug=True) 能成功完成基本的增删改查：
打开里面具体看看（我就列举一个啊，我懒）：
写这一部分是简单梳理一下部分模块的逻辑和方法，这次实例中呢，除了服务模块（methods_face）外，其他模块都比较好理解，那我们就来梳理一下这个模块吧~
导出需要的模块：
from flask import Flask, Blueprint from flask_restplus import Api, Resource, fields from database import mysql_pool import logger_error from flask_restplus import reqparse 注册蓝图并设置一下需要的配置信息：
#注册蓝图 api_v1 = Blueprint('api', __name__) #配置蓝图信息 api = Api(api_v1, version='1.0', title='TodoMVC API', description='A simple TodoMVC API', ) #设置api名字 ns = api.namespace('todos', description='TODO operations') #设置api的model的样式 todo = api.model('Todo', { 'id': fields.Integer(readOnly=True, description='The task unique identifier'),#整型字段 'author': fields.String(required=True),#String字段 'age': fields.Integer(required=True), 'current_address': fields.String(required=True), 'times_awards': fields.Integer(required=True) }) #使用一个解析器先定义好我们需要的参数，方便后面去解析这些参数 parser = reqparse.RequestParser(bundle_errors=True) parser.add_argument('id', type=int, required=True) parser.add_argument('author', type=str, required=True) parser.add_argument('age', type=int, required=True) parser.add_argument('current_address', type=str, required=True) parser.add_argument('times_awards', type=int, required=True) 加入数据库连接池：
POOL = mysql_pool() 接口来了！
@ns.route('/create_todo') class Todo(Resource): @ns.doc('create_todo') @ns.expect(todo) @ns.marshal_with(todo, code=201) def post(self): '''创建一个新的task''' # return SQL.create(api.payload), 201 # data = json.loads(api.payload) todo = {} id = todo['id'] = api.payload['id'] author = todo['author'] = api.payload['author'] age = todo['age'] = api.payload['age'] current_address = todo['current_address'] = api.payload['current_address'] times_awards = todo['times_awards'] = api.payload['times_awards'] conn = POOL.connection() cursor = conn.cursor() sql_insert = "INSERT INTO author (id,author,age,current_address,times_awards) VALUES (%s,%s,%s,%s,%s)" cursor.execute(sql_insert, (id, author, age, current_address, times_awards)) conn.commit() conn.close() return todo @ns.route('/get_todo/<int:id>') class TodoSimple(Resource): @api.doc('get_todo') @ns.marshal_with(todo) def get(self, id): '''获取id指定的todo项''' try: conn = POOL.connection() cursor = conn.cursor() sql_select = "select * from author where id = %s" if cursor.execute(sql_select, id): info = cursor.fetchall() conn.close() jsondata = [] for row in info: result = {} result["id"] = row[0] result["author"] = row[1] result["age"] = row[2] result["current_address"] = row[3] result["times_awards"] = row[4] jsondata.append(result) return jsondata else: api.abort(404, "Todo {} doesn't exist".format(id)) return todo except: logger_error() pass @ns.route('/delete_todo/<int:id>') class TodoSimple(Resource): @ns.doc('delete_todo') @ns.response(204, 'Todo deleted') def delete(self, id): '''根据id删除对应的task''' try: conn = POOL.connection() cursor = conn.cursor() sql_drop = "DELETE FROM author WHERE id = %s" cursor.execute(sql_drop, id) conn.commit() conn.close() return '', 204 except: logger_error() return '', 404 @ns.route('/update_todo/<int:id>') class TodoSimple(Resource): @ns.doc('update_todo') @ns.expect(todo) @ns.marshal_with(todo) def put(self, id): '''更新id指定的task''' try: todo = {} if id == api.payload['id']: todo['id'] = id author = todo['author'] = api.payload['author'] age = todo['age'] = api.payload['age'] current_address = todo['current_address'] = api.payload['current_address'] times_awards = todo['times_awards'] = api.payload['times_awards'] conn = POOL.connection() cursor = conn.cursor() sql_update = "UPDATE author SET author=%s,age=%s,current_address=%s,times_awards=%s WHERE id = %s" cursor.execute(sql_update, (author, age, current_address, times_awards, id)) conn.commit() conn.close() return todo except: pass logger_error() return '', 404注意点：
1.api.payload来传入参数
2.简单介绍一下swagger文档，就是@api.doc,@api.expect,@api.marshal_with,@api.route
(1)api.doc()装饰器允许您在文档中包含额外的信息
(2)expect()装饰器允许您指定预期的输入字段。它接受一个可选的布尔参数validate，该参数指示是否应该验证负载
(3)这个装饰器的工作方式类似于原始marshal_with()装饰器，不同之处在于它记录了这些方法。可选参数代码允许指定预期的HTTP状态代码(默认为200)。可选参数as_list允许您指定对象是否作为列表返回。
(4)可以使用Api.route()的doc参数提供类范围的文档。此参数接受与Api.doc()装饰器相同的值
如果大家只是想知道怎么简单的运用，那么看我代码+这一部分的说明应该就可以，如果大家想深入了解一下原理和更高级的运行，可以去看swagger官方文档，如果大家想要中文版的，可以去下载一下我上传的"python 学习模块.rar"资源（放心，免费的，不会让你们破费的hh，给博客点个赞就成）
flask中有很多相互包容的包，在有对应包情况下尽量避免自己去手动创建。这次实例中使用的数据库连接是我查询资料仿照连接池来写的，这一部分应该去使用和flask框架兼容性更好的sqlalchemy模块去连接数据库。
工作方式类似于原始marshal_with()装饰器，不同之处在于它记录了这些方法。可选参数代码允许指定预期的HTTP状态代码(默认为200)。可选参数as_list允许您指定对象是否作为列表返回。
(4)可以使用Api.route()的doc参数提供类范围的文档。此参数接受与Api.doc()装饰器相同的值
如果大家只是想知道怎么简单的运用，那么看我代码+这一部分的说明应该就可以，如果大家想深入了解一下原理和更高级的运行，可以去看swagger官方文档，如果大家想要中文版的，可以去下载一下我上传的"python 学习模块.rar"资源（放心，免费的，不会让你们破费的hh，给博客点个赞就成）
flask中有很多相互包容的包，在有对应包情况下尽量避免自己去手动创建。这次实例中使用的数据库连接是我查询资料仿照连接池来写的，这一部分应该去使用和flask框架兼容性更好的sqlalchemy模块去连接数据库。
写在最后：好久没写博客了 o(╥﹏╥)o ，不过没事看的人也不多，我会把pycharm项目压缩上传成资源的。我会尽快更新下一部分——flask-restplus(数据库连接基于sqlalchemy)
