python - 使用 MySQL.connector 和 Twisted Python 执行多个查询

Question

我最近有一个 Python 2.7x 项目，我需要使用 mysql.connector 在一个查询中执行多个用分号分隔的语句。这在 post. 中有很好的解释。 .

但是，我当前的项目需要将 mysql.connector 与 Twisted 一起使用，这意味着使用 Twisted 优秀的 enterprise.adbapi 模块来使我的新阻塞数据库连接成为非阻塞。

config = {"user": username, "password": password, "host": hostname, "database": database_name, "raise_on_warnings": True} cp = adbapi.ConnectionPool("mysql.connector", **config)

我的测试语句定义如下。我很抱歉它们是一个有点无聊的例子，但我知道我期望的结果，并且它应该足以验证我是否获得了多个语句的结果。

    statement1 = "SELECT * FROM queue WHERE id = 27;"
    statement2 = "SELECT * FROM order WHERE id = 1;"
    statement_list = [statement1, statement2]
    statements = " ".join(statement_list)

当我现在尝试执行 ConnectionPool 方法 .runQuery() 时，问题就出现了

    def _print_result(result):
        if result:
            print("this is a result")
            print(result)
        else:
            print("no result")
        reactor.stop()

    d = cp.runQuery(statements, multi=True)
    d.addBoth(_print_result)

这给我带来了以下结果:

this is a result [Failure instance: Traceback: : No result set to fetch from.

如何使用 Twisted 的 adbapi 模块来获取我知道的结果？

Answer 1

因此，事实证明，当使用 adbapi.ConnectionPool.runQuery() 时，默认行为是将数据库询问的结果发送到cursor.fetchall() 方法。但是，当使用 mysql.connector 时，即使没有扭曲，这也不起作用。相反，我们需要迭代结果集，并对结果集的每个成员调用 fetchall()。

所以，我解决这个问题的方法是使用以下子类。

from twisted.enterprise import adbapi

class NEWadbapiConnectionPool(adbapi.ConnectionPool):

def __init__(self, dbapiName, *connargs, **connkw):
    adbapi.ConnectionPool.__init__(self, dbapiName, *connargs, **connkw)

def runMultiQuery(self, *args, **kw):
    return self.runInteraction(self._runMultiQuery, *args, **kw)

def _runMultiQuery(self, trans, *args, **kw):
    result = trans.execute(*args, **kw)
    result_list = []
    for item in result:
        if item.with_rows:
            result_list.append(item.fetchall())
    return result_list

所以现在我创建以下内容:

    def _print_result(result):
        if result:
            print("this is a result")
            print(result)
        else:
            print("no result")
        reactor.stop()

    cp = NEWadbapiConnectionPool("mysql.connector", **config)
    d = cp.runMultiQuery(statements, multi=True)
    d.addBoth(_print_result)

并获取每个语句的结果列表。

我希望其他人发现这很有用。