Python SQLite3 查询结果返回字典的常见错误及完整解决方案

import sqlite3
conn = sqlite3.connect('database.db')
cu = conn.cursor()
cu.execute("select id, username, email from user")
result = cu.fetchall()
print(result)
# 输出: [(1, 'alice', 'alice@example.com'), (2, 'bob', 'bob@example.com')]
# 只有值，没有字段名

import sqlite3
from flask import jsonify
def userSearch():
    # 连接数据库
    conn = handleDbConnection()
    # 关键步骤1: 设置 row_factory
    conn.row_factory = sqlite3.Row
    # 创建游标并执行查询
    cu = conn.cursor()
    cu.execute("select * from user")
    # 获取查询结果
    listAll = cu.fetchall()
    # 关键步骤2: 将 Row 对象转换为字典
    listAll = [dict(row) for row in listAll]
    # 现在 listAll 是字典列表
    # [{'id': 1, 'username': 'alice', 'email': 'alice@example.com'}, ...]
    return jsonify({
        "code": 200,
        "data": {
            "list": listAll,
            "total": len(listAll)
        }
    })

row = cu.fetchone()
# 可以通过索引访问
print(row[0])  # 第一列的值
# 可以通过列名访问
print(row['username'])  # username 字段的值
# 可以转换为字典
print(dict(row))  # {'id': 1, 'username': 'alice', ...}

[dict(row) for row in listAll]

[表达式 for 变量 in 可迭代对象]

result = []
for row in listAll:
    result.append(dict(row))
listAll = result

# 转换前（Row 对象列表）
listAll = [
    <sqlite3.Row object at 0x...>,
    <sqlite3.Row object at 0x...>
]
# 转换后（字典列表）
listAll = [
    {'id': 1, 'username': 'alice', 'email': 'alice@example.com'},
    {'id': 2, 'username': 'bob', 'email': 'bob@example.com'}
]

# 示例1: 生成平方数列表
numbers = [1, 2, 3, 4, 5]
squares = [x**2 for x in numbers]
print(squares)  # [1, 4, 9, 16, 25]
# 示例2: 转换为大写
names = ['alice', 'bob', 'charlie']
upper_names = [name.upper() for name in names]
print(upper_names)  # ['ALICE', 'BOB', 'CHARLIE']
# 示例3: 提取字典的某个字段
users = [
    {'name': 'alice', 'age': 25},
    {'name': 'bob', 'age': 30}
]
names = [user['name'] for user in users]
print(names)  # ['alice', 'bob']

# 语法：[表达式 for 变量 in 可迭代对象 if 条件]
# 示例: 只转换偶数
numbers = [1, 2, 3, 4, 5, 6]
even_squares = [x**2 for x in numbers if x % 2 == 0]
print(even_squares)  # [4, 16, 36]
# 在 SQLite 场景中的应用
listAll = [dict(row) for row in listAll if row['age'] > 18]
# 只转换年龄大于18的记录

# 示例: 扁平化二维列表
matrix = [[1, 2], [3, 4], [5, 6]]
flat = [num for row in matrix for num in row]
print(flat)  # [1, 2, 3, 4, 5, 6]

# 错误示例
listAll = cu.fetchall()  # Row 对象列表
return jsonify({"data": listAll})
# TypeError: Object of type Row is not JSON serializable
# 正确示例
listAll = [dict(row) for row in cu.fetchall()]
return jsonify({"data": listAll})
# 成功序列化为 JSON

# 使用字典
user = {'id': 1, 'username': 'alice', 'email': 'alice@example.com'}
print(user['username'])  # 清晰明了
# 使用 tuple（默认）
user = (1, 'alice', 'alice@example.com')
print(user[1])  # 需要记住索引位置，容易出错

import sqlite3
from typing import List, Dict, Any
def query_to_dict(conn: sqlite3.Connection, sql: str, params: tuple = ()) -> List[Dict[str, Any]]:
    """
    执行 SQL 查询并返回字典列表
    Args:
        conn: 数据库连接对象
        sql: SQL 查询语句
        params: 查询参数（可选）
    Returns:
        字典列表，每个字典代表一行数据
    """
    # 设置 row_factory
    conn.row_factory = sqlite3.Row
    cu = conn.cursor()
    # 执行查询
    cu.execute(sql, params)
    # 转换为字典列表
    result = [dict(row) for row in cu.fetchall()]
    # 关闭游标
    cu.close()
    return result
# 使用示例
conn = sqlite3.connect('database.db')
users = query_to_dict(conn, "SELECT * FROM user WHERE age > ?", (18,))
print(users)
# [{'id': 1, 'username': 'alice', 'age': 25}, ...]

def userSearch(pageNum: int = 1, pageSize: int = 10) -> dict:
    """
    用户查询（带分页）
    Args:
        pageNum: 页码（从1开始）
        pageSize: 每页数量
    Returns:
        包含列表和分页信息的字典
    """
    conn = handleDbConnection()
    conn.row_factory = sqlite3.Row
    cu = conn.cursor()
    # 查询总数
    cu.execute("SELECT COUNT(*) as total FROM user")
    total = cu.fetchone()['total']
    # 计算偏移量
    offset = (pageNum - 1) * pageSize
    # 查询分页数据
    cu.execute("SELECT * FROM user LIMIT ? OFFSET ?", (pageSize, offset))
    listAll = [dict(row) for row in cu.fetchall()]
    cu.close()
    conn.close()
    return {
        "code": 200,
        "data": {
            "list": listAll,
            "total": total,
            "pageNum": pageNum,
            "pageSize": pageSize,
            "totalPages": (total + pageSize - 1) // pageSize
        },
        "message": "成功"
    }

import time
# 方法1: 使用列表推导式（推荐）
start = time.time()
result1 = [dict(row) for row in listAll]
print(f"列表推导式耗时: {time.time() - start:.4f}秒")
# 方法2: 使用 map 函数
start = time.time()
result2 = list(map(dict, listAll))
print(f"map 函数耗时: {time.time() - start:.4f}秒")
# 方法3: 使用传统循环
start = time.time()
result3 = []
for row in listAll:
    result3.append(dict(row))
print(f"传统循环耗时: {time.time() - start:.4f}秒")

# 转换时修改字段名
listAll = [
    {
        'userId': row['id'],
        'userName': row['username'],
        'userEmail': row['email']
    }
    for row in cu.fetchall()
]

# 添加额外的计算字段
listAll = [
    {
        **dict(row),  # 展开原始字段
        'fullName': f"{row['first_name']} {row['last_name']}",
        'isAdmin': row['role'] == 'admin'
    }
    for row in cu.fetchall()
]

# 只保留非空字段
listAll = [
    {k: v for k, v in dict(row).items() if v is not None}
    for row in cu.fetchall()
]

# 错误
conn = sqlite3.connect('db.sqlite')
cu = conn.cursor()
cu.execute("SELECT * FROM user")
listAll = [dict(row) for row in cu.fetchall()]
# TypeError: cannot convert dictionary update sequence element #0 to a sequence

# 正确
conn.row_factory = sqlite3.Row  # 必须添加这一行

# 错误：在 cursor 之后设置
cu = conn.cursor()
conn.row_factory = sqlite3.Row  # 太晚了！

# 正确：在 cursor 之前设置
conn.row_factory = sqlite3.Row
cu = conn.cursor()

# 错误
return jsonify({"data": cu.fetchall()})
# TypeError: Object of type Row is not JSON serializable

# 正确
return jsonify({"data": [dict(row) for row in cu.fetchall()]})