在python sqlite中将结果行Map到namedtuple

c7rzv4ha  于 2023-10-23  发布在  SQLite
关注(0)|答案(5)|浏览(162)

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

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

文档建议的Map代码如下:

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约束来设置name字段,以便只保证一个结果。

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

另一种方法可以使用fetchall()[0]而不使用unique约束来保证只得到一个结果。
我的问题是,哪一种是处理这个问题的最好和最常用的方法,我应该总是使用fetchall来维护一个公共接口,让数据库管理唯一性逻辑吗?或者我应该像方法1那样显式地创建一个列表?有没有更简单的方法来完成这项任务?

oxf4rvwz

oxf4rvwz1#

有一个更简单的方法!Sqlite3为用户提供了一种定义“行工厂”的方法。这些行工厂接受游标和元组行,并可以返回任何类型的对象。
一旦将行工厂设置为

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)

有关如何定义namedtuple工厂的另一个示例,请参见this post
顺便说一下,如果你设置

conn.row_factory = sqlite3.Row

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

pnwntuvh

pnwntuvh2#

一个改进的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()
gr8qqesn

gr8qqesn3#

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

当所有查询结果都具有相同的字段时,这些工厂函数非常有用。
示例如下:

  1. namedtuple_factory = make_row_factory(namedtuple)
  2. import dataclass
    row_factory = make_row_factory(dataclass.make_dataclass)
  3. 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

命名元组:

from collections import namedtuple
Row = namedtuple("Row", "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
lnvxswe2

lnvxswe24#

我演示了如何从SQL查询中获取结果框架并将其转换为namedtuple列表。我没有动态地将数组列绑定到nametuple名称,不确定这是否可行。

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')]
nr9pn0ug

nr9pn0ug5#

我认为最好使用for language in map(LanguageRecord._make, c.fetchall()[:1]):,因为它可能会导致fetchall()[0]的IndexError。
如果你只需要一个结果,而查询中已经有了“WHERE”。据我所知,查询应该返回一行。早期优化是邪恶的。:)

相关问题