基於python3.x，使用Tornado中的torndb模塊操作數據庫

目前Tornado中的torndb模塊是不支持python3.x，所以需要修改部分torndb源碼即可正常使用

1、開發環境介紹

操作系統：win8（64位），python版本：python3.6（32位），IDE：pycharm

2、安裝torndb（這里使用pip進行安裝）

pip install torndb

3、源碼修改

• 修改MySQLdb，torndb是依賴於MySQLdb實現的對MySQL數據庫操作，但是python3中不支持MySQLdb，而是使用pymysql，所以需要將源碼中使用MySQLdb的地方修改為pymysql。

1）修改導入模塊

import pymysql.connections
import pymysql.converters
import pymysql.cursors
# import MySQLdb.constants
# import MySQLdb.converters
# import MySQLdb.cursors

 2）修改連接mysql的方式

    def reconnect(self):
        """Closes the existing database connection and re-opens it."""
        self.close()
        self._db = pymysql.connect(**self._db_args)# MySQLdb.connect(**self._db_args)
        self._db.autocommit(True)

3）修改連接參數，以及遍歷字段類型時所使用的列表增加元素（python3使用append進行元素的添加，而不是使用加號）

 

# if MySQLdb is not None:
if pymysql is not None:
    # Fix the access conversions to properly recognize unicode/binary
    FIELD_TYPE = pymysql.connections.FIELD_TYPE # MySQLdb.constants.FIELD_TYPE
    FLAG = pymysql.constants.FLAG# MySQLdb.constants.FLAG
    CONVERSIONS = copy.copy (pymysql.converters.conversions)# (MySQLdb.converters.conversions)

    field_types = [FIELD_TYPE.BLOB, FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING]
    if 'VARCHAR' in vars(FIELD_TYPE):
        field_types.append(FIELD_TYPE.VARCHAR)

    for field_type in field_types:
        # CONVERSIONS[field_type] = [(FLAG.BINARY, str)] + CONVERSIONS[field_type]
        CONVERSIONS[field_type] = [(FLAG.BINARY, str)].append(CONVERSIONS[field_type])

    # Alias some common MySQL exceptions
    IntegrityError = pymysql.IntegrityError# MySQLdb.IntegrityError
    OperationalError = pymysql.OperationalError# MySQLdb.OperationalError

 

• 修改連接超時時間，在torndb初始化方法中設置，需要傳遞給pymysql

  def __init__(self, host, database, user=None, password=None,
                   max_idle_time=7 * 3600, connect_timeout=10,# 設置連接超時時間，時間是秒
                   time_zone="+0:00", charset = "utf8", sql_mode="TRADITIONAL"):

• 修改查詢方法中的迭代方法（將izip改為zip_longest）

    def query(self, query, *parameters, **kwparameters):
        """Returns a row list for the given query and parameters."""
        cursor = self._cursor()
        try:
            self._execute(cursor, query, parameters, kwparameters)
            column_names = [d[0] for d in cursor.description]
            return [Row(itertools.zip_longest(column_names, row)) for row in cursor]
        finally:
            cursor.close()

 

　　

 

4、測試使用

• 數據庫

 

• 源碼

class IndexHandler(RequestHandler):
    def get(self, *args, **kwargs):
        # get返回的是字典對象
        ret = self.application.db.get('select title from houses where id = 2')
        self.write(ret['title'])

if __name__ == '__main__':
    options.parse_command_line()
    app = Application([
        (r'/', IndexHandler),
    ], debug=True)
    # 建立數據庫連接
    app.db = torndb.Connection(
        host='127.0.0.1',
        database='demo',
        user='you_ruser',
        password='your_password'
    )
    http_server = httpserver.HTTPServer(app)
    http_server.listen(options.port)
    ioloop.IOLoop.current().start()

• 訪問

• 經過測試，增刪改查等數據庫操作能夠正常使用，就不一一展示了