Python Peewee execute_sql() 示例

Question

我在我的项目中使用Peewee模块作为ORM。

我阅读了整个文档，但是没有明确的示例说明如何处理db.execute_sql()的结果。

我追踪了代码，只能找到db.execute_sql()返回游标。

是否有人知道如何处理游标，例如遍历它并从复杂的选择语句中获取结果。

更新：我刚刚在peewee文件夹中找到了以下源代码，它应该可以帮助我解决这个问题。

class QueryResultWrapper(object):
    """
    提供对原始查询结果的迭代器，此外还执行两个操作:
    -将数据库行转换为Python表示
    -确保多次迭代不会导致多个查询
    """
    def __init__(self, model, cursor, meta=None):
        self.model = model
        self.cursor = cursor

        self.__ct = 0
        self.__idx = 0
self._result_cache = []
        self._populated = False
        self._initialized = False

        if meta is not None:
            self.column_meta, self.join_meta = meta
        else:
            self.column_meta = self.join_meta = None

    def __iter__(self):
        self.__idx = 0
if not self._populated:
            return self
        else:
            return iter(self._result_cache)
def process_row(self, row):
        return row
def iterate(self):
        row = self.cursor.fetchone()
        if not row:
            self._populated = True
            raise StopIteration
        elif not self._initialized:
            self.initialize(self.cursor.description)
            self._initialized = True
        return self.process_row(row)
def iterator(self):
        while True:
            yield self.iterate()
def next(self):
        if self.__idx < len(self):
            res = self[self.__idx]
            self.__idx += 1
            return res
        elif self.__idx == len(self) and not self._populated:
            return self.iterate()
        else:
            raise StopIteration

- user1187968

---

返回的对象类型是什么？运行 sql_execute() 并打印结果以查看其类型。 - Ayush

Answer 1

Peewee返回一个游标。然后你可以使用db-api 2来迭代它:

cursor = db.execute_sql('select * from tweets;')
for row in cursor.fetchall():
    print(row)

cursor = db.execute_sql('select count(*) from tweets;')
res = cursor.fetchone()
print('Total: ', res[0])

Docs: Database.execute_sql

- coleifer

---

有没有一种获取返回列名映射字典的方法？ - KJW

@SagarShah 它必须映射到具有给定表名和字段的类，像 show table status from db_name 这样的查询怎么样？它映射到无表。 - TomSawyer

@KJW 请尝试使用 cursor.description 获取列名：col_names = [col[0] for col in cursor.description] res = [dict(zip(col_names, row)) for row in cursor.fetchall()] - Mart Van de Ven