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

Question

我正在使用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在数据库中使用约束设置名称字段,以便仅对一个结果进行garantize.

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

另一种方法可以在fetchall()[0]没有unique约束的情况下使用,仅仅对一个结果进行garantize.

我的问题是哪种方法是处理这个问题的最好和最常用的方法,我应该总是使用fetchall维护一个公共接口并让数据库管理唯一性逻辑吗？或者我应该如方法1中那样明确地创建列表？有没有更简单的方法来完成这项任务？

Answer 1

有一个更简单的方法!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工厂的另一个示例,请参阅此文章.

顺便说一句,如果你设置

conn.row_factory = sqlite3.Row

然后行返回为dicts,其键是表的列名.因此,不要使用类似的东西访问namedtuple的部分,而row.creation_date只需使用内置sqlite3.Row行工厂并访问等效项row['creation_date'].

Answer 2

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