基于tornado---异步并发接口

import reuqests

POST = requests.post # post请求方式

def db_query(sql, method='query'):
    db_api = 'http://127.0.0.1:8000/db/api'
    db_base = "dbbase"
    db_ret = POST(db_api, data=json_encode({'method': method, 'sql': sql, 'dbbase': db_base, 'pwd': 'password123'}))
    if db_ret.status_code == 200:
        if json_decode(db_ret.text)['status'] == 'True':
            db_ret_data = json_decode(db_ret.text)['data']
            return {'status': 'ok', 'data': db_ret_data, 'err': ''}
        else:
            return {'status': 'False', 'data': [], 'err': json_decode(db_ret.text)['data']}
    else:
        return {'status': 'False', 'data': [], 'err': 'connect error'}

# coding=utf8
import MySQLdb
import MySQLdb.cursors
import tornado
import tornado.ioloop
import tornado.web
import tornado.gen
from tornado.concurrent import run_on_executor
from concurrent.futures import ThreadPoolExecutor


class DB_CONFIG:
    config = {
        'dbbase': ('authentic', 'password'), # 数据库user、password
    }


class DB:
    def __init__(self, dbbase):
        user, password = DB_CONFIG.config.get(dbbase, (None, None))
        if user == None or password == None:
            raise Exception('KeyError', dbbase)
        db_host = '192.168.xx.xx'
        db_port = 1234

        self.db = MySQLdb.connect(db_host, user, password, dbbase, port=db_port, cursorclass=MySQLdb.cursors.DictCursor)

        self.cursor = self.db.cursor()

    def query(self, sql):
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def commit(self, sql):
        try:
            self.cursor.execute(sql)
            self.db.commit()
            return {'status': True, 'data': ''}
        except Exception as e:
            self.db.rollback()
            return {'status': False, 'data': e}

    def close(self):
        self.db.close()


class ServiceHandler(tornado.web.RequestHandler):
    executor = ThreadPoolExecutor(900)  # 必须定义一个executor的属性，然后run_on_executor装饰器才会游泳。

    @run_on_executor  # 线程内运行；query函数被run_on_executor包裹(语法糖)，将该函数的执行传递给线程池executor的线程执行，优化了处理耗时性任务，以致达到不阻塞主线程的效果。
    def query(self, dbname, method, sql):
        db = DB(dbname)
        ret = ''
        if method == 'query':
            ret = db.query(sql)
        elif method == 'commit':
            ret = db.commit(sql)
        db.close()
        return ret

    @tornado.web.asynchronous  # 保持长连接，直到处理后返回
    @tornado.gen.coroutine  # 异步、协程处理；增加并发量
    def post(self):
        data = tornado.escape.json_decode(self.request.body)  # 获取参数，json.loads()解码
        r = {'status': '', 'data': ''}
        if not data.get('pwd', None):
            r['status'], r['data'] = ('False', 'not password')
        elif not data.get('dbbase', None):
            r['status'], r['data'] = ('False', 'not DB select')
        else:
            if data['pwd'] != 'password123)': # 接口的密码认证
                r = {'status': 'False', 'data': 'password error'}
            elif data['method'] == 'query':

                d = yield self.query(data['dbbase'], 'query', data['sql'])
                r = {'status': 'True', 'data': d}
            elif data['method'] == 'commit':

                db_r = yield self.query(data['dbbase'], 'commit', data['sql'])
                if db_r['status']:
                    r = {'status': 'True', 'data': 'commit sucessful'}
                else:
                    r = {'status': 'False', 'data': db_r['data']}
            else:
                r = {'status': 'False', 'data': 'method Invaild'}
        self.write(tornado.escape.json_encode(r))  # 写入返回信息写入response
        self.finish()  # 结束服务

    def get(self):
        return self.post()


if __name__ == "__main__":
    application = tornado.web.Application([
        (r"/db/api", ServiceHandler),  # 路由映射
    ])
    application.listen(8000)  # 监听端口
    tornado.ioloop.IOLoop.instance().start()  # 启动服务

 sql = 'select vid from video where status=1 group by vid ORDER BY num ASC limit 100'

db_ret = db_query(sql, 'query') # 其中query是数据库操作的方法 query为查询/commit 为insert/update/delete等数据库修改操作时使用