在python sqlite中将结果行映射到namedtuple

Posted 2023-02-16

【中文标题】在python sqlite中将结果行映射到namedtuple

【英文标题】：Mapping result rows to namedtuple in python sqlite

【发布时间】：2013-04-26 11:47:25

【问题描述】：

我正在使用 sqlite3 的 python api，我有一个用于存储语言的小表，其中包含 id、name 和 creation_date 字段。我正在尝试按照文档的建议将原始查询结果映射到namedtuple，这样我就可以以更易读的方式管理行，所以这是我的namedtuple。

LanguageRecord = namedtuple('LanguageRecord', 'id, name, creation_date')

文档建议的映射代码如下：

for language in map(LanguageRecord._make, c.fetchall()):
  # do something with languages

当我想返回一组语言时这很好，但在这种情况下我只想 检索一种语言：

c.execute('SELECT * FROM language WHERE name=?', (name,))

所以我的第一次尝试是这样的：

language = map(LanguageRecord._make, c.fetchone())

此代码不起作用，因为fetchone() 返回一个元组而不是一个包含一个元组的列表， 所以map 函数尝试为每个元组字段创建三个namedtuples。

我解决这个问题的第一种方法是显式创建一个列表并将元组结果附加到它上面，类似于：

languages = []
languages.append(c.fetchone())
for language in map(LanguageRecord._make, languages):
  # do something with language

我的第二种方法是使用fetchall()，尽管我只想要一条记录。我可以设置 数据库中带有unique 约束的名称字段，以便只保证一个结果。

for language in map(LanguageRecord._make, c.fetchall()):
  # do something with languages

另一种方法是使用 fetchall()[0] 而不使用 unique 约束来保证一个结果。

我的问题是处理这个问题的最好和最常用的方法是什么，我应该总是使用fetchall 来维护一个通用接口并让数据库管理唯一性逻辑吗？还是我应该像方法1一样明确创建一个列表？有没有更简单的方法来完成这项任务？

【问题讨论】：

你也可以遍历数据库游标，不需要获取所有记录，除非你想，所以代码可以重写为map(LanguageRecord._make, c)。

【参考方案1】：

有一个更简单的方法！ Sqlite3 提供了一种方式供用户定义"row factories"。这些行工厂采用游标和元组行，可以返回它想要的任何类型的对象。

一旦你设置了行工厂

con.row_factory = my_row_factory

那么游标返回的行将是my_row_factory 应用于元组行的结果。例如，

import sqlite3
import collections

LanguageRecord = collections.namedtuple('LanguageRecord', 'id name creation_date')
def namedtuple_factory(cursor, row):
    return LanguageRecord(*row)

con = sqlite3.connect(":memory:")
con.row_factory = namedtuple_factory
cur = con.cursor()
cur.execute("select 1,2,3")
print(cur.fetchone())

产量

LanguageRecord(id=1, name=2, creation_date=3)

---

有关如何定义命名元组工厂的另一个示例，请参阅this post。

---

顺便说一句，如果你设置了

conn.row_factory = sqlite3.Row

然后行作为字典返回，其键是表的列名。因此，您可以使用内置的sqlite3.Row 行工厂并使用row['creation_date'] 访问等价物，而不是使用row.creation_date 之类的东西访问namedtuple 的一部分。

【讨论】：

sqlite3.Row 不是真正的字典，因为它没有实现 .get() 或 .__contains__()。

【参考方案2】：

我认为更好用 for language in map(LanguageRecord._make, c.fetchall()[:1]): 因为它可能会导致 fetchall()[0] 出现 IndexError。

如果您需要一个结果并且查询中已经存在“WHERE”。据我了解，查询应该返回一行。早期优化是邪恶的。 :)

【参考方案3】：

改进后的row_factory其实是这样的，可以重复用于各种查询：

from collections import namedtuple

def namedtuple_factory(cursor, row):
    """Returns sqlite rows as named tuples."""
    fields = [col[0] for col in cursor.description]
    Row = namedtuple("Row", fields)
    return Row(*row)

conn = sqlite3.connect(":memory:")
conn.row_factory = namedtuple_factory
cur = con.cursor()

【讨论】：

在 named_factory 函数中设置光标和行的内容。我没有看到这些值被传递给函数

@GoldenLion row 是具有单行值的元组。 Row(*row) 是将值传递给将返回的命名元组的位置。

row_factory 指向 namedtuple_factory 函数。 con.cursor 会调用点函数吗？

row_factory 由sqlite3 模块在内部使用。见docs.python.org/3/library/…

谢谢，我意识到这是模式

【参考方案4】：

namedtuple上方还有一个row_factory：

from collection import namedtuple

def namedtuple_factory(cursor, row, cls=[None]):
    rf = cls[0]
    if rf is None:
        fields = [col[0] for col in cursor.description]
        cls[0] = namedtuple("Row", fields)
        return cls[0](*row)
    return rf(*row)

可以进一步概括以使用其他类工厂：

def make_row_factory(cls_factory, **kw):
    def row_factory(cursor, row, cls=[None]):
        rf = cls[0]
        if rf is None:
            fields = [col[0] for col in cursor.description]
            cls[0] = cls_factory("Row", fields, **kw)
            return cls[0](*row)
        return rf(*row)
    return row_factory

这些工厂函数对于所有查询结果都具有相同字段的情况很有用。

例子：

namedtuple_factory = make_row_factory(namedtuple)

import dataclass

row_factory = make_row_factory(dataclass.make_dataclass)

pip3 install recordclass

import recordclass

row_factory = make_row_factory(recordclass.make_dataclass, fast_new=True)

这里有一些性能计数器来比较不同的方式（debian linux、64 位、python 3.9）。

创建测试数据库的脚本：

N = 1000000
conn = sqlite3.connect('example.db')
c = conn.cursor()
c.execute('''CREATE TABLE test
             (id int, x real, y real, p int, q int)''')
gen = ((i, random(), random(), randint(0,N), randint(0,N)) for i in range(N))
c.executemany("INSERT INTO test VALUES (?,?,?,?,?)", gen)
conn.commit()
conn.close()

默认值：

conn = sqlite3.connect('example.db')
c = conn.cursor()
%time res = [row for row in c.execute("SELECT id,x,y,p,q FROM test")]
conn.close()
print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 971 ms, sys: 92.1 ms, total: 1.06 s
Wall time: 1.06 s
80 Mb

sqlite3.Row：

conn = sqlite3.connect('example.db')
conn.row_factory = sqlite3.Row
c = conn.cursor()
%time res = [row for row in c.execute("SELECT id,x,y,p,q FROM test")]
conn.close()
# print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 1.11 s, sys: 80.1 ms, total: 1.19 s
Wall time: 1.19 s

命名元组：

从集合导入命名元组 行 = 命名元组（“行”，“id x y p q”） conn = sqlite3.connect('example.db') c = conn.cursor() %time res = [Row(*row) for row in c.execute("SELECT id,x,y,p,q FROM test")] conn.close() print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 1.89 s, sys: 71.8 ms, total: 1.96 s
Wall time: 1.96 s
80 Mb

基于命名元组的行工厂：

conn = sqlite3.connect('example.db')
conn.row_factory = make_row_factory(namedtuple)
c = conn.cursor()
%time res = [row for row in c.execute("SELECT id,x,y,p,q FROM test")]
conn.close()
print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 1.93 s, sys: 116 ms, total: 2.05 s
Wall time: 2.05 s
80 Mb

记录类：

from recordclass import make_dataclass
Row = make_dataclass("Row", "id x y p q", fast_new=True)
conn = sqlite3.connect('example.db')
c = conn.cursor()
%time res = [Row(*row) for row in c.execute("SELECT id,x,y,p,q FROM test")]
conn.close()
print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 1 s, sys: 72.2 ms, total: 1.08 s
Wall time: 1.07 s
56 Mb

基于记录类的行工厂：

conn = sqlite3.connect('example.db')
conn.row_factory = make_row_factory(make_dataclass, fast_new=True)
c = conn.cursor()
%time res = [row for row in c.execute("SELECT id,x,y,p,q FROM test")]
conn.close()
print(N * sys.getsizeof(res[0]) // 1000000, 'Mb')

CPU times: user 1.11 s, sys: 76.2 ms, total: 1.19 s
Wall time: 1.19 s
56 Mb

【参考方案5】：

我演示了如何从 sql 查询中获取结果数据帧并将其转换为命名元组列表。我没有将数据框列动态绑定到名称元组名称，不确定这是否可行。

LanguageRecord=namedtuple('Generic',['id','name','creation_date'])
def map_to_language_record(row):
    return LanguageRecord(row.id, row.name, row.creation_date)

df=pd.DataFrame('id':[1,2,3],'name':['bob','dick','jane'],'creation_date': 
  ['1/1/2021','1/2/2021','1/3/2021'])

languages = list(map(map_to_language_record, df.itertuples()))
print(languages)

输出：

[Generic(id=1, name='bob', creation_date='1/1/2021'), Generic(id=2, name='dick', creation_date='1/2/2021'), Generic(id=3, name='jane', creation_date='1/3/2021')]