Python sqlite3.Cursor.row_factory 属性
上次修改时间：2025 年 4 月 15 日
本综合指南探讨了 Python 的 sqlite3.Cursor.row_factory 属性，该属性控制如何从查询返回 SQLite 行。我们将介绍基本用法、内置工厂、自定义实现和实际示例。
基本定义
sqlite3.Cursor 的 row_factory 属性决定了从数据库查询返回结果行时的格式。默认情况下，行以元组形式返回。
主要特点：它接受一个可调用对象来处理原始行，可以在连接或游标级别设置，并支持对列的命名访问。这个强大的特性提高了代码的可读性和可维护性。
默认行工厂（元组）
默认情况下，行以元组形式返回，值可以通过索引位置访问。此示例演示了默认行为。
import sqlite3
with sqlite3.connect(':memory:') as conn:
conn.execute('CREATE TABLE users(id INTEGER, name TEXT)')
conn.execute('INSERT INTO users VALUES(1, "Alice")')
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')
row = cursor.fetchone()
print(row) # (1, 'Alice')
print(row[0]) # 1 (access by index)
print(row[1]) # 'Alice'
这显示了基于元组的默认行格式。虽然简单，但通过索引访问列会使代码更难阅读和维护，尤其是在列很多的情况下。
默认设置效率很高，但缺乏对列的命名访问，而其他行工厂在这方面很有用。
使用 sqlite3.Row 工厂
内置的 sqlite3.Row 工厂提供命名列访问和其他有用的功能。这是大多数用例的推荐方法。
import sqlite3
with sqlite3.connect(':memory:') as conn:
conn.row_factory = sqlite3.Row
conn.execute('CREATE TABLE books(id INTEGER, title TEXT, author TEXT)')
conn.execute('INSERT INTO books VALUES(1, "Python Basics", "John Doe")')
cursor = conn.cursor()
cursor.execute('SELECT * FROM books')
row = cursor.fetchone()
print(row['title']) # 'Python Basics'
print(row['author']) # 'John Doe'
print(row.keys()) # ['id', 'title', 'author']
sqlite3.Row 提供了几个优点：按名称访问列，不区分大小写的名称，以及类似字典的方法。它比自定义字典工厂的内存效率更高。
当您需要对列进行命名访问，而又不想为每行创建完整的字典时，此工厂是理想的选择。
自定义字典工厂
您可以创建一个自定义工厂，将行作为字典返回。这提供了行格式化的最大灵活性。
import sqlite3
def dict_factory(cursor, row):
return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
with sqlite3.connect(':memory:') as conn:
conn.row_factory = dict_factory
conn.execute('CREATE TABLE products(id INTEGER, name TEXT, price REAL)')
conn.execute('INSERT INTO products VALUES(1, "Laptop", 999.99)')
cursor = conn.cursor()
cursor.execute('SELECT * FROM products')
product = cursor.fetchone()
print(product) # {'id': 1, 'name': 'Laptop', 'price': 999.99}
print(product['name']) # 'Laptop'
此自定义工厂将每一行转换为以列名作为键的字典。cursor.description 属性提供列元数据。
字典行便于 JSON 序列化，或者当您需要完整的字典功能时。但是，它们比 sqlite3.Row 使用更多的内存。
命名元组工厂
为了在可读性和性能之间取得平衡，您可以使用命名元组作为行工厂。这提供了命名访问，而开销比字典小。
import sqlite3
from collections import namedtuple
def namedtuple_factory(cursor, row):
fields = [col[0] for col in cursor.description]
Row = namedtuple('Row', fields)
return Row(*row)
with sqlite3.connect(':memory:') as conn:
conn.row_factory = namedtuple_factory
conn.execute('CREATE TABLE employees(id INTEGER, name TEXT, department TEXT)')
conn.execute('INSERT INTO employees VALUES(1, "Alice", "Engineering")')
cursor = conn.cursor()
cursor.execute('SELECT * FROM employees')
employee = cursor.fetchone()
print(employee) # Row(id=1, name='Alice', department='Engineering')
print(employee.name) # 'Alice' (attribute access)
print(employee[0]) # 1 (still supports index access)
此工厂为每一行创建轻量级的命名元组对象。命名元组是不可变的并且内存高效，同时提供点符号访问。
当您需要命名访问和类似元组的行为时，尤其是在对性能敏感的应用程序中，这种方法非常出色。
连接级别与游标级别工厂
行工厂可以在连接级别或游标级别设置。此示例显示了如何在不同级别使用不同的工厂。
import sqlite3
def dict_factory(cursor, row):
return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
with sqlite3.connect(':memory:') as conn:
# Set default factory for all cursors
conn.row_factory = sqlite3.Row
conn.execute('CREATE TABLE test(id INTEGER, data TEXT)')
conn.execute('INSERT INTO test VALUES(1, "Connection level")')
conn.execute('INSERT INTO test VALUES(2, "Cursor level")')
# Default cursor uses connection's factory
cursor1 = conn.cursor()
cursor1.execute('SELECT * FROM test WHERE id = 1')
row1 = cursor1.fetchone()
print(row1['data']) # 'Connection level'
# Custom cursor overrides connection factory
cursor2 = conn.cursor()
cursor2.row_factory = dict_factory
cursor2.execute('SELECT * FROM test WHERE id = 2')
row2 = cursor2.fetchone()
print(row2['data']) # 'Cursor level'
连接级别的工厂默认应用于所有游标，而游标级别的工厂会覆盖连接设置。这允许灵活的行格式化。
对于跨查询的一致性，请使用连接级别，而当您需要特定查询的不同格式时，请使用游标级别。
JSON 序列化工厂
对于 Web 应用程序，您可能需要格式化为易于 JSON 序列化的行。此示例显示了一个自定义的 JSON 友好工厂。
import sqlite3
import json
from datetime import datetime
def json_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
value = row[idx]
if isinstance(value, datetime):
value = value.isoformat()
d[col[0]] = value
return d
with sqlite3.connect(':memory:') as conn:
conn.row_factory = json_factory
conn.execute('CREATE TABLE events(id INTEGER, name TEXT, created_at TEXT)')
conn.execute('INSERT INTO events VALUES(1, "Launch", ?)',
(datetime.now().isoformat(),))
cursor = conn.cursor()
cursor.execute('SELECT * FROM events')
event = cursor.fetchone()
print(json.dumps(event)) # {"id": 1, "name": "Launch", "created_at": "2025-04-15T..."}
此工厂确保所有值都是 JSON 可序列化的，自动将日期转换为 ISO 格式字符串。它非常适合返回数据库结果的 API。
自定义工厂可以处理任何特殊的格式化需求，使其成为数据转换的强大工具。
类型转换工厂
行工厂还可以执行类型转换。此示例自动将特定列转换为 Python 类型。
import sqlite3
import json
def convert_types(value):
if value.startswith('{') and value.endswith('}'):
try:
return json.loads(value)
except:
return value
return value
def type_aware_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
value = row[idx]
if isinstance(value, str):
value = convert_types(value)
d[col[0]] = value
return d
with sqlite3.connect(':memory:') as conn:
conn.row_factory = type_aware_factory
conn.execute('CREATE TABLE config(id INTEGER, settings TEXT)')
conn.execute('INSERT INTO config VALUES(1, \'{"theme":"dark"}\')')
cursor = conn.cursor()
cursor.execute('SELECT * FROM config')
config = cursor.fetchone()
print(config['settings']['theme']) # 'dark' (automatically parsed JSON)
此工厂自动检测并将 JSON 字符串转换为 Python 字典。您可以扩展它来处理日期、小数或其他专门的格式。
类型转换工厂通过在行级别而不是在应用程序逻辑中处理转换来减少样板代码。
最佳实践
- 首选 sqlite3.Row： 对于大多数情况，它提供了最佳平衡
- 在连接级别设置： 为了跨查询的一致性
- 谨慎使用自定义工厂： 它们会增加开销
- 考虑内存使用： 字典比元组使用更多的内存
- 记录您的选择： 尤其是在使用自定义工厂时
资料来源
作者
列出所有 Python 教程。