row_factory上面还有一个namedtuple：
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.行：
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 xyp 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