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()]})