Python SQLite：从表中选择数据深度解析

weixin_42107409

1280人浏览 · 2025-09-18 09:01:08

weixin_42107409 · 2025-09-18 09:01:08 发布

引言：数据库的洞察之窗——数据查询
SQLite SELECT 语句基础
- 2.1 选择所有列 (SELECT *)
- 2.2 选择特定列 (SELECT column1, column2)
- 2.3 别名 (AS)
- 2.4 唯一值 (DISTINCT)
筛选数据：WHERE 子句详解
- 3.1 比较运算符 (=, !=, <, >, <=, >=)
- 3.2 逻辑运算符 (AND, OR, NOT)
- 3.3 范围查询 (BETWEEN)
- 3.4 列表匹配 (IN, NOT IN)
- 3.5 模式匹配 (LIKE, GLOB)
  - 3.5.1 LIKE 操作符 (%, _)
  - 3.5.2 GLOB 操作符 (类似正则表达式)
- 3.6 空值判断 (IS NULL, IS NOT NULL)
排序、限制与分页
- 4.1 排序结果 (ORDER BY)
  - 4.1.1 升序 (ASC) 与降序 (DESC)
  - 4.1.2 多列排序
- 4.2 限制结果数量 (LIMIT)
- 4.3 结果偏移 (OFFSET) 与分页
在 Python sqlite3 中查询数据
- 5.1 建立连接与获取游标 (with 语句)
- 5.2 执行 SELECT 语句：参数绑定 (? 或 :name)
  - 5.2.1 位置参数绑定
  - 5.2.2 命名参数绑定
- 5.3 获取查询结果
  - 5.3.1 cursor.fetchone()
  - 5.3.2 cursor.fetchmany(size)
  - 5.3.3 cursor.fetchall()
  - 5.3.4 直接迭代游标
- 5.4 结果集增强：sqlite3.Row
- 5.5 自动类型转换 (detect_types)
高级查询概念 (简述)
- 6.1 聚合函数 (COUNT, SUM, AVG, MIN, MAX)
- 6.2 分组数据 (GROUP BY)
- 6.3 过滤分组 (HAVING)
- 6.4 表连接 (JOIN)
  - 6.4.1 INNER JOIN
  - 6.4.2 LEFT JOIN
  - 6.4.3 SQLite 对 RIGHT JOIN 和 FULL JOIN 的限制
- 6.5 子查询 (Subqueries)
错误处理
- 7.1 sqlite3.OperationalError
最佳实践与注意事项
- 8.1 始终使用参数绑定进行查询 (防止 SQL 注入)
- 8.2 使用 with 语句管理数据库连接和游标
- 8.3 仅选择必需的列 (SELECT column1, column2 而非 SELECT *)
- 合理利用 LIMIT 和 OFFSET 进行分页
- 考虑索引以优化查询性能 (简要提及)
- 外键约束的启用 (PRAGMA foreign_keys = ON;)
- 理解并利用 sqlite3.Row 和 detect_types
- ALTER TABLE 限制与常见变通方案
- 安全警告：避免动态 DDL/DML 中的用户输入
综合代码示例
- 9.1 表结构和初始数据
- 9.2 各种 SELECT 语句演示 (筛选、排序、分页、聚合、连接)
总结

1. 引言：数据库的洞察之窗——数据查询

在完成了数据库的创建和数据的插入之后，**查询数据（Select Data）**是数据库操作中最为常见和重要的环节。SELECT 语句允许我们从一个或多个表中检索信息，它是获取数据库洞察力的核心工具。无论是生成报告、显示用户资料，还是进行数据分析，都离不开强大的 SELECT 语句。

本指南将深入探讨 SQLite SELECT 语句的各种用法，包括基本的列选择、复杂的筛选条件、排序、分页以及如何通过 Python 的 sqlite3 模块高效且安全地执行这些查询，并处理查询结果。

2. SQLite `SELECT` 语句基础

2.1 选择所有列 (`SELECT *`)

检索表中所有列的数据。

SELECT * FROM table_name;

示例:

SELECT * FROM users;

2.2 选择特定列 (`SELECT column1, column2`)

只检索您感兴趣的特定列的数据。这是一种推荐的做法，因为它能减少网络传输和内存使用。

SELECT column1, column2 FROM table_name;

示例:

SELECT name, email FROM users;

2.3 别名 (`AS`)

可以为列或表指定一个临时的别名，以提高查询的可读性或简化引用。AS 关键字是可选的。

SELECT column_name AS alias_name FROM table_name;
SELECT t.column_name FROM table_name AS t;

示例:

SELECT first_name AS "First Name", email AS UserEmail FROM customers;

2.4 唯一值 (`DISTINCT`)

使用 DISTINCT 关键字可以从结果集中消除重复的行。它作用于 SELECT 后面的所有列组合。

SELECT DISTINCT column1 FROM table_name;
SELECT DISTINCT column1, column2 FROM table_name;

示例:

SELECT DISTINCT registration_date FROM customers;

3. 筛选数据：`WHERE` 子句详解

WHERE 子句用于根据指定的条件过滤行，只有满足条件的行才会被包含在结果集中。

3.1 比较运算符 (`=`, `!=`, `<`, `>`, `<=`, `>=`)

SELECT * FROM products WHERE price > 100;
SELECT * FROM customers WHERE first_name = 'Alice';
SELECT * FROM orders WHERE total_amount <= 50;

3.2 逻辑运算符 (`AND`, `OR`, `NOT`)

AND: 两个条件都为真。
OR: 至少一个条件为真。
NOT: 条件为假。

SELECT * FROM products WHERE price > 50 AND stock < 100;
SELECT * FROM customers WHERE first_name = 'Alice' OR first_name = 'Bob';
SELECT * FROM orders WHERE NOT is_completed;

3.3 范围查询 (`BETWEEN`)

用于检查一个值是否在指定范围内（包含边界值）。

SELECT * FROM products WHERE price BETWEEN 10.00 AND 50.00; -- 价格在 10.00 到 50.00 之间 (包含)

等价于：WHERE price >= 10.00 AND price <= 50.00。

3.4 列表匹配 (`IN`, `NOT IN`)

用于检查一个值是否在一组给定值中。

SELECT * FROM customers WHERE first_name IN ('Alice', 'Charlie', 'Eve');
SELECT * FROM products WHERE name NOT IN ('Laptop', 'Mouse');

3.5 模式匹配 (`LIKE`, `GLOB`)

3.5.1 `LIKE` 操作符 (`%`, `_`)

用于字符串的模式匹配。

%: 匹配零个、一个或多个字符。
_: 匹配任何单个字符。

SELECT * FROM customers WHERE email LIKE '%@example.com'; -- 邮箱以 @example.com 结尾
SELECT * FROM products WHERE name LIKE 'L_ptop';          -- 匹配 'Laptop', 'Luptop' 等

3.5.2 `GLOB` 操作符 (类似正则表达式)

GLOB 操作符的功能与 LIKE 相似，但它使用 Unix 文件名匹配的通配符规则，区分大小写（LIKE 默认不区分）。

*: 匹配零个或多个字符。
?: 匹配任何单个字符。
[]: 匹配方括号内的任何单个字符。
[^]: 匹配不在方括号内的任何单个字符。

SELECT * FROM customers WHERE first_name GLOB 'A*'; -- 匹配以 'A' 开头的名字
SELECT * FROM products WHERE name GLOB 'M?use';    -- 匹配 'Mouse'

3.6 空值判断 (`IS NULL`, `IS NOT NULL`)

用于检查列是否包含 NULL 值。不能使用 = NULL 或 != NULL。

SELECT * FROM users WHERE email IS NULL;
SELECT * FROM users WHERE email IS NOT NULL;

4. 排序、限制与分页

4.1 排序结果 (`ORDER BY`)

ORDER BY 子句用于对查询结果进行排序。

4.1.1 升序 (`ASC`) 与降序 (`DESC`)

ASC: 升序 (默认)。
DESC: 降序。

SELECT name, price FROM products ORDER BY price ASC;   -- 按价格升序
SELECT name, stock FROM products ORDER BY stock DESC; -- 按库存降序

4.1.2 多列排序

可以指定多个列进行排序。排序的优先级从左到右。

SELECT first_name, last_name, registration_date FROM customers
ORDER BY registration_date DESC, last_name ASC; -- 先按注册日期降序，日期相同则按姓氏升序

4.2 限制结果数量 (`LIMIT`)

LIMIT 子句用于限制结果集中返回的行数。

SELECT * FROM products ORDER BY price DESC LIMIT 3; -- 获取价格最高的 3 个产品

4.3 结果偏移 (`OFFSET`) 与分页

OFFSET 子句与 LIMIT 结合使用，用于跳过结果集中的前 N 行，实现分页功能。

-- 获取第 2 页的数据，每页 5 条 (跳过前 5 条，取接下来的 5 条)
SELECT * FROM products ORDER BY name ASC LIMIT 5 OFFSET 5;

LIMIT 5 OFFSET 0: 第一页 (前 5 条)
LIMIT 5 OFFSET 5: 第二页 (跳过 5 条，取接下来的 5 条)
LIMIT 5 OFFSET 10: 第三页 (跳过 10 条，取接下来的 5 条)

5. 在 Python `sqlite3` 中查询数据

5.1 建立连接与获取游标 (`with` 语句)

import sqlite3
import os
import datetime

db_name = 'select_db.db'
if os.path.exists(db_name):
    os.remove(db_name)

# 推荐使用 with 语句管理连接和事务
# 启用 detect_types 以自动转换日期时间和布尔值
# 设置 row_factory 使得可以通过列名访问结果
conn = sqlite3.connect(db_name, detect_types=sqlite3.PARSE_DECLTYPES)
conn.row_factory = sqlite3.Row # 可以通过 row['column_name'] 访问数据

# 在 with 块中使用连接和游标
with conn: # conn 作为一个上下文管理器，会自动 commit/rollback 和 close
    cursor = conn.cursor()
    cursor.execute("PRAGMA foreign_keys = ON;") # 启用外键

    # 创建并填充示例表 (为了演示 SELECT，我们先快速创建和插入一些数据)
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS customers (
            customer_id INTEGER PRIMARY KEY AUTOINCREMENT,
            first_name TEXT NOT NULL,
            last_name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            registration_date DATE DEFAULT CURRENT_DATE
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            product_id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            price REAL CHECK(price > 0) DEFAULT 0.0,
            stock INTEGER DEFAULT 0 CHECK(stock >= 0)
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS orders (
            order_id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id INTEGER NOT NULL,
            order_date DATETIME DEFAULT CURRENT_TIMESTAMP,
            total_amount REAL,
            is_completed BOOLEAN DEFAULT FALSE,
            FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
                ON DELETE CASCADE
        )
    ''')

    # 插入一些示例数据
    cursor.executemany("INSERT INTO customers (first_name, last_name, email, registration_date) VALUES (?, ?, ?, ?)", [
        ("Alice", "Smith", "alice@example.com", datetime.date(2023, 1, 10)),
        ("Bob", "Johnson", "bob@example.com", datetime.date(2023, 1, 15)),
        ("Charlie", "Brown", "charlie@example.com", datetime.date(2023, 2, 20)),
        ("David", "Lee", "david@example.com", datetime.date(2023, 3, 5)),
        ("Eve", "Wang", "eve@example.com", datetime.date(2023, 3, 10)),
        ("Frank", "Green", "frank@example.com", datetime.date(2023, 4, 1))
    ])
    cursor.executemany("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", [
        ("Laptop", 1200.50, 20),
        ("Mouse", 25.00, 150),
        ("Keyboard", 75.99, 80),
        ("Monitor", 300.00, 30),
        ("Webcam", 49.99, 100),
        ("Desk Lamp", 35.00, 60),
        ("USB Hub", 15.00, 200)
    ])
    
    # 获取客户ID
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'alice@example.com'")
    alice_id = cursor.fetchone()['customer_id']
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'bob@example.com'")
    bob_id = cursor.fetchone()['customer_id']
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'charlie@example.com'")
    charlie_id = cursor.fetchone()['customer_id']

    cursor.executemany("INSERT INTO orders (customer_id, order_date, total_amount, is_completed) VALUES (?, ?, ?, ?)", [
        (alice_id, datetime.datetime(2023, 1, 20, 10, 30), 1250.00, False),
        (bob_id, datetime.datetime(2023, 3, 1, 14, 0), 50.00, True),
        (alice_id, datetime.datetime(2023, 2, 5, 9, 0), 200.00, True),
        (charlie_id, datetime.datetime(2023, 4, 10, 11, 45), 800.00, False),
        (bob_id, datetime.datetime(2023, 3, 10, 16, 20), 15.00, True),
        (alice_id, datetime.datetime(2023, 4, 2, 13, 0), 300.00, False)
    ])
    print("数据库和初始数据已准备好。\n")

5.2 执行 `SELECT` 语句：参数绑定 (`?` 或 `:name`)

与 INSERT 类似，始终使用参数绑定来传递 WHERE 子句或其他需要动态值的 SQL 语句。这可以防止 SQL 注入攻击，并确保正确的数据类型处理。

5.2.1 位置参数绑定

使用问号 ? 作为占位符，值作为元组传递给 execute() 方法的第二个参数。

search_name = 'Laptop'
cursor.execute("SELECT * FROM products WHERE name = ?", (search_name,)) # 注意，即使只有一个参数，也需要是元组
product = cursor.fetchone()
print(f"查询产品 '{search_name}': {product['name']}, 价格: {product['price']}")

5.2.2 命名参数绑定

使用 :name 或 $name 或 @name 作为占位符，值作为字典传递。

min_price = 100.0
max_price = 500.0
cursor.execute("SELECT name, price FROM products WHERE price BETWEEN :min_p AND :max_p ORDER BY price DESC",
               {'min_p': min_price, 'max_p': max_price})
print(f"\n查询价格在 {min_price} 到 {max_price} 之间的产品:")
for p in cursor.fetchall():
    print(f"  {p['name']}: ${p['price']}")

5.3 获取查询结果

游标对象提供多种方法来检索查询结果。

5.3.1 `cursor.fetchone()`

获取查询结果集的下一行。如果没有更多行，则返回 None。

cursor.execute("SELECT * FROM customers WHERE customer_id = ?", (alice_id,))
alice = cursor.fetchone()
print(f"\nAlice 的信息: {alice['first_name']} {alice['last_name']}, 注册日期: {alice['registration_date']}")

5.3.2 `cursor.fetchmany(size)`

获取查询结果集的接下来 size 行。如果没有足够的行，则返回所有剩余的行。如果没有更多行，则返回空列表。

cursor.execute("SELECT * FROM products ORDER BY product_id")
first_two_products = cursor.fetchmany(2)
print("\n前两件产品:")
for p in first_two_products:
    print(f"  {p['name']}, 价格: {p['price']}")

next_three_products = cursor.fetchmany(3)
print("\n接下来的三件产品:")
for p in next_three_products:
    print(f"  {p['name']}, 价格: {p['price']}")

5.3.3 `cursor.fetchall()`

获取查询结果集的所有（剩余）行，并作为一个列表返回。如果没有更多行，则返回空列表。

cursor.execute("SELECT name, stock FROM products WHERE stock < 100")
low_stock_products = cursor.fetchall()
print("\n库存低于 100 的产品:")
for p in low_stock_products:
    print(f"  {p['name']} (库存: {p['stock']})")

5.3.4 直接迭代游标

游标对象本身就是可迭代的，可以直接在其上循环以逐行获取结果。这对于处理大量结果集非常有效，因为它不需要一次性将所有行加载到内存中。

print("\n所有客户的邮箱 (迭代游标):")
cursor.execute("SELECT email FROM customers")
for row in cursor: # row 将是 sqlite3.Row 或元组，取决于 row_factory
    print(f"  {row['email']}")

5.4 结果集增强：`sqlite3.Row`

默认情况下，cursor.fetchone()、fetchmany() 和 fetchall() 返回的行是元组。这意味着您只能通过索引访问列 (例如 row[0])。

通过设置 conn.row_factory = sqlite3.Row，返回的行将是 sqlite3.Row 对象。sqlite3.Row 允许您：

通过列名访问数据: row['column_name'] (推荐，提高可读性)。
通过索引访问数据: row[0]。
表现得像元组，同时又像字典。

上述所有示例都已默认使用 sqlite3.Row，并通过列名访问数据。

5.5 自动类型转换 (`detect_types`) (续)

我们之前的 connect 调用已经包含了 detect_types=sqlite3.PARSE_DECLTYPES，因此在查询 registration_date 和 order_date 时，它们将自动作为 datetime.date 或 datetime.datetime 对象返回，is_completed 将作为 bool 返回。

# 再次查询 Alice 的信息，验证日期类型自动转换
cursor.execute("SELECT customer_id, first_name, registration_date FROM customers WHERE customer_id = ?", (alice_id,))
alice_info = cursor.fetchone()
print(f"\nAlice 的信息 (ID: {alice_info['customer_id']}):")
print(f"  姓名: {alice_info['first_name']}")
print(f"  注册日期: {alice_info['registration_date']} (类型: {type(alice_info['registration_date'])})") # 预期 datetime.date

# 查询订单信息，验证日期时间和布尔类型自动转换
cursor.execute("SELECT order_id, order_date, is_completed FROM orders WHERE customer_id = ? LIMIT 1", (alice_id,))
alice_order = cursor.fetchone()
print(f"\nAlice 的一个订单信息:")
print(f"  订单ID: {alice_order['order_id']}")
print(f"  下单日期: {alice_order['order_date']} (类型: {type(alice_order['order_date'])})") # 预期 datetime.datetime
print(f"  是否完成: {alice_order['is_completed']} (类型: {type(alice_order['is_completed'])})") # 预期 bool

6. 高级查询概念 (简述)

SQL 的 SELECT 语句远不止于简单的选择和过滤，它还包括强大的聚合、分组和连接功能。

6.1 聚合函数 (`COUNT`, `SUM`, `AVG`, `MIN`, `MAX`)

用于对一组行执行计算并返回单个值。

COUNT(*) / COUNT(column): 计算行数。
SUM(column): 计算列值的总和。
AVG(column): 计算列值的平均值。
MIN(column): 获取列的最小值。
MAX(column): 获取列的最大值。

SELECT COUNT(*) FROM customers; -- 客户总数
SELECT SUM(total_amount) FROM orders; -- 订单总金额
SELECT AVG(price) FROM products; -- 产品平均价格

6.2 分组数据 (`GROUP BY`)

GROUP BY 子句用于将具有相同值的行分组，通常与聚合函数一起使用。

SELECT customer_id, COUNT(order_id) AS total_orders FROM orders GROUP BY customer_id; -- 每个客户的订单数量

6.3 过滤分组 (`HAVING`)

HAVING 子句用于过滤 GROUP BY 结果中的组，其作用类似于 WHERE，但作用于分组后的聚合结果。

SELECT customer_id, COUNT(order_id) AS total_orders
FROM orders
GROUP BY customer_id
HAVING total_orders >= 2; -- 查找订单数量大于或等于 2 的客户

6.4 表连接 (`JOIN`)

JOIN 子句用于根据两个或多个表之间的相关列，将这些表中的行组合起来。

6.4.1 `INNER JOIN`

只返回两个表中都匹配的行。

SELECT c.first_name, c.email, o.order_id, o.total_amount
FROM customers AS c
INNER JOIN orders AS o ON c.customer_id = o.customer_id; -- 查找所有客户及其对应的订单

6.4.2 `LEFT JOIN` (或 `LEFT OUTER JOIN`)

返回左表的所有行，以及右表中匹配的行。如果右表中没有匹配项，则右表的列将为 NULL。

SELECT c.first_name, c.email, o.order_id
FROM customers AS c
LEFT JOIN orders AS o ON c.customer_id = o.customer_id; -- 查找所有客户，以及他们可能有的订单

6.4.3 SQLite 对 `RIGHT JOIN` 和 `FULL JOIN` 的限制

SQLite 不直接支持 RIGHT JOIN 和 FULL OUTER JOIN。

RIGHT JOIN 可以通过交换表的位置并使用 LEFT JOIN 来模拟。
FULL OUTER JOIN 通常需要通过 LEFT JOIN 和 UNION ALL 结合 RIGHT JOIN 的模拟来实现。

6.5 子查询 (Subqueries)

子查询（或内查询）是嵌套在另一个 SQL 查询中的查询。它通常用于作为主查询的条件或数据源。

-- 查找下过订单的客户姓名
SELECT first_name, last_name FROM customers
WHERE customer_id IN (SELECT DISTINCT customer_id FROM orders);

7. 错误处理

在执行 SELECT 查询时，可能会遇到 sqlite3.OperationalError。

7.1 `sqlite3.OperationalError`

当 SQL 语法错误、引用的表或列不存在、数据库文件被锁定或损坏等操作性错误发生时，会抛出 sqlite3.OperationalError。

try:
    cursor.execute("SELECT non_existent_column FROM non_existent_table")
    results = cursor.fetchall()
except sqlite3.OperationalError as e:
    print(f"\n捕获到 OperationalError: {e}")
    print("查询失败，可能是表或列名错误。")

8. 最佳实践与注意事项

8.1 始终使用参数绑定进行查询 (防止 SQL 注入)

无论是 INSERT 还是 SELECT，只要 SQL 语句中包含来自外部（尤其是用户）的变量值，就必须使用参数绑定。这是数据库安全的基石。

8.2 使用 `with` 语句管理数据库连接和游标

这确保了连接和游标在不再需要时会被正确关闭，并且在发生错误时会自动回滚事务（对于 SELECT 语句，虽然不修改数据，但它确保了连接的正确关闭）。

8.3 仅选择必需的列 (`SELECT column1, column2` 而非 `SELECT *`)

避免使用 SELECT *，除非您确实需要所有列。显式选择所需的列可以：

提高性能: 减少从磁盘读取的数据量和网络传输量。
降低内存消耗: Python 端处理的数据量更小。
增强代码可读性: 明确查询的目的。
未来兼容性: 当表结构发生变化（如添加新列）时，使用 SELECT * 可能会导致不必要的兼容性问题或错误。

8.4 合理利用 `LIMIT` 和 `OFFSET` 进行分页

对于大型结果集，一次性获取所有数据可能会导致内存不足或响应缓慢。使用 LIMIT 和 OFFSET 实现分页，分批加载数据，可以提高应用程序的响应性和用户体验。

8.5 考虑索引以优化查询性能 (简要提及)

对于经常用于 WHERE 子句、JOIN 条件或 ORDER BY 子句的列，创建索引可以显著提高查询速度。虽然本篇主要关注 SELECT 语句，但性能优化是数据库操作不可或缺的一部分。

8.6 外键约束的启用 (`PRAGMA foreign_keys = ON;`)

尽管在 SELECT 操作中外键约束不会直接“生效”（因为它不修改数据），但它确保了数据库中数据的完整性。在查询时，您可以依赖这种完整性进行数据关联和业务逻辑判断。因此，始终在每个连接中启用它是一个良好的习惯。

8.7 理解并利用 `sqlite3.Row` 和 `detect_types`

sqlite3.Row: 使得查询结果可以通过列名访问，大大提高了代码的可读性和可维护性。
detect_types: 简化了 Python 与 SQLite 之间日期时间、布尔值等复杂数据类型的转换，减少了手动解析的错误和代码量。

8.8 `ALTER TABLE` 限制与常见变通方案

（复用前两篇的提醒）SQLite 对 ALTER TABLE 的支持相对有限。例如，它不能直接删除列，也不能直接修改列的数据类型或约束（除了少数例外）。对于更复杂的模式更改（例如，删除列、修改列类型、重新排序列），常见的变通方法是：

创建具有所需新结构的新表。
将旧表中的数据复制到新表（只复制兼容的列）。
删除旧表。
将新表重命名为旧表的名称。
理解这些限制对于规划数据库模式演进至关重要。

8.9 安全警告：避免动态 DDL/DML 中的用户输入

（复用前两篇的提醒）尽管本篇主要讨论 SELECT，但如果您的应用程序需要根据用户输入动态地生成 SQL 查询（例如，动态选择列名、表名或复杂的 WHERE 子句），请务必对这些输入进行严格验证和白名单过滤。表名、列名和 SQL 关键字不能通过参数绑定传递。直接拼接未经净化的用户输入到 SQL 语句中可能会导致严重的 SQL 注入漏洞。在可能的情况下，应避免使用用户提供的名称来动态生成 SQL。

9. 综合代码示例

此示例将基于之前创建的数据库和数据，演示各种 SELECT 语句的使用，包括筛选、排序、分页、聚合和连接。

# 假设我们已经执行了 5.1 节中的数据库初始化和数据插入代码
# 接续之前的 with conn 块，或重新建立连接并插入数据 (此处为了独立演示，重新连接并插入数据)

import sqlite3
import os
import datetime

db_name = 'comprehensive_select_db.db'
if os.path.exists(db_name):
    os.remove(db_name)

with sqlite3.connect(db_name, detect_types=sqlite3.PARSE_DECLTYPES) as conn:
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute("PRAGMA foreign_keys = ON;")

    # --- 准备数据 (与 5.1 节相同，为了代码的完整性和独立性) ---
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS customers (
            customer_id INTEGER PRIMARY KEY AUTOINCREMENT,
            first_name TEXT NOT NULL,
            last_name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            registration_date DATE DEFAULT CURRENT_DATE
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            product_id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            price REAL CHECK(price > 0) DEFAULT 0.0,
            stock INTEGER DEFAULT 0 CHECK(stock >= 0)
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS orders (
            order_id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id INTEGER NOT NULL,
            order_date DATETIME DEFAULT CURRENT_TIMESTAMP,
            total_amount REAL,
            is_completed BOOLEAN DEFAULT FALSE,
            FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
                ON DELETE CASCADE
        )
    ''')
    cursor.executemany("INSERT INTO customers (first_name, last_name, email, registration_date) VALUES (?, ?, ?, ?)", [
        ("Alice", "Smith", "alice@example.com", datetime.date(2023, 1, 10)),
        ("Bob", "Johnson", "bob@example.com", datetime.date(2023, 1, 15)),
        ("Charlie", "Brown", "charlie@example.com", datetime.date(2023, 2, 20)),
        ("David", "Lee", "david@example.com", datetime.date(2023, 3, 5)),
        ("Eve", "Wang", "eve@example.com", datetime.date(2023, 3, 10)),
        ("Frank", "Green", "frank@example.com", datetime.date(2023, 4, 1))
    ])
    cursor.executemany("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", [
        ("Laptop", 1200.50, 20),
        ("Mouse", 25.00, 150),
        ("Keyboard", 75.99, 80),
        ("Monitor", 300.00, 30),
        ("Webcam", 49.99, 100),
        ("Desk Lamp", 35.00, 60),
        ("USB Hub", 15.00, 200)
    ])
    
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'alice@example.com'")
    alice_id = cursor.fetchone()['customer_id']
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'bob@example.com'")
    bob_id = cursor.fetchone()['customer_id']
    cursor.execute("SELECT customer_id FROM customers WHERE email = 'charlie@example.com'")
    charlie_id = cursor.fetchone()['customer_id']

    cursor.executemany("INSERT INTO orders (customer_id, order_date, total_amount, is_completed) VALUES (?, ?, ?, ?)", [
        (alice_id, datetime.datetime(2023, 1, 20, 10, 30), 1250.00, False),
        (bob_id, datetime.datetime(2023, 3, 1, 14, 0), 50.00, True),
        (alice_id, datetime.datetime(2023, 2, 5, 9, 0), 200.00, True),
        (charlie_id, datetime.datetime(2023, 4, 10, 11, 45), 800.00, False),
        (bob_id, datetime.datetime(2023, 3, 10, 16, 20), 15.00, True),
        (alice_id, datetime.datetime(2023, 4, 2, 13, 0), 300.00, False)
    ])
    print("数据库和初始数据已准备好。\n")

    # --- 演示各种 SELECT 查询 ---

    print("--- 1. 选择所有客户 (SELECT *) ---")
    cursor.execute("SELECT * FROM customers")
    for row in cursor.fetchall():
        print(f"  ID: {row['customer_id']}, 姓名: {row['first_name']} {row['last_name']}, 邮箱: {row['email']}")

    print("\n--- 2. 选择特定列和别名 (SELECT column AS alias) ---")
    cursor.execute("SELECT first_name, email AS ContactEmail FROM customers WHERE customer_id = ?", (alice_id,))
    alice = cursor.fetchone()
    print(f"  Alice 的邮箱: {alice['ContactEmail']}")

    print("\n--- 3. 筛选数据 (WHERE, AND, OR, LIKE) ---")
    cursor.execute("SELECT name, price FROM products WHERE price > ? AND name LIKE ?", (100, '%or%'))
    print("  价格大于 100 且名称包含 'or' 的产品:")
    for product in cursor.fetchall():
        print(f"    {product['name']} (价格: ${product['price']})")

    print("\n--- 4. 排序和限制 (ORDER BY, LIMIT) ---")
    cursor.execute("SELECT name, stock FROM products ORDER BY stock DESC LIMIT 3")
    print("  库存前 3 的产品:")
    for product in cursor.fetchall():
        print(f"    {product['name']} (库存: {product['stock']})")

    print("\n--- 5. 分页查询 (LIMIT, OFFSET) ---")
    page_size = 2
    for page_num in range(3): # 查询 3 页
        offset = page_num * page_size
        cursor.execute("SELECT first_name, last_name FROM customers ORDER BY customer_id LIMIT ? OFFSET ?", (page_size, offset))
        customers_page = cursor.fetchall()
        if not customers_page:
            break
        print(f"  客户列表 - 第 {page_num + 1} 页:")
        for customer in customers_page:
            print(f"    {customer['first_name']} {customer['last_name']}")

    print("\n--- 6. 聚合函数 (COUNT, SUM, AVG) ---")
    cursor.execute("SELECT COUNT(*) AS total_customers FROM customers")
    print(f"  总客户数: {cursor.fetchone()['total_customers']}")

    cursor.execute("SELECT SUM(total_amount) AS total_revenue FROM orders")
    print(f"  总收入: ${cursor.fetchone()['total_revenue']:.2f}")

    print("\n--- 7. 分组和过滤分组 (GROUP BY, HAVING) ---")
    cursor.execute('''
        SELECT customer_id, COUNT(order_id) AS order_count, SUM(total_amount) AS total_spent
        FROM orders
        GROUP BY customer_id
        HAVING order_count >= ?
        ORDER BY total_spent DESC
    ''', (2,))
    print("  下过 2 个或更多订单的客户及其消费总额:")
    for customer_stats in cursor.fetchall():
        # 获取客户姓名
        cursor.execute("SELECT first_name, last_name FROM customers WHERE customer_id = ?", (customer_stats['customer_id'],))
        customer_name = cursor.fetchone()
        print(f"    {customer_name['first_name']} {customer_name['last_name']} (订单数: {customer_stats['order_count']}, 总消费: ${customer_stats['total_spent']:.2f})")

    print("\n--- 8. 表连接 (INNER JOIN) ---")
    cursor.execute('''
        SELECT c.first_name, c.last_name, o.order_id, o.order_date, o.total_amount
        FROM customers AS c
        INNER JOIN orders AS o ON c.customer_id = o.customer_id
        WHERE o.is_completed = ?
        ORDER BY o.order_date DESC
    ''', (True,)) # 查找已完成的订单及其客户信息
    print("  已完成订单的客户和订单详情:")
    for order_detail in cursor.fetchall():
        print(f"    客户: {order_detail['first_name']} {order_detail['last_name']}, 订单ID: {order_detail['order_id']}, 日期: {order_detail['order_date'].strftime('%Y-%m-%d %H:%M')}, 金额: ${order_detail['total_amount']:.2f}")

    print("\n--- 9. 子查询 (Subquery) ---")
    cursor.execute('''
        SELECT first_name, last_name FROM customers
        WHERE customer_id IN (
            SELECT customer_id FROM orders WHERE total_amount > ?
        )
    ''', (500,)) # 查找有超过 500 美元订单的客户
    print("  有高价值订单的客户:")
    for customer in cursor.fetchall():
        print(f"    {customer['first_name']} {customer['last_name']}")

print(f"\n所有查询演示完成，数据库 '{db_name}' 已更新。")

10. 总结

为您详细介绍了 Python SQLite 中从表中选择数据的过程。

核心要点回顾:

掌握 SELECT 语句的各种基本和高级用法，包括选择特定列、使用别名、过滤 (WHERE)、排序 (ORDER BY)、限制 (LIMIT)、分页 (OFFSET)、聚合 (COUNT, SUM, AVG, MIN, MAX)、分组 (GROUP BY, HAVING) 和表连接 (JOIN)。
在 Python 中，始终使用参数绑定 (? 或 :name) 来安全地传递查询参数，防止 SQL 注入。
理解并高效利用 cursor.fetchone(), fetchmany(), fetchall() 以及直接迭代游标来检索查询结果。
通过 conn.row_factory = sqlite3.Row 增强结果集，允许通过列名访问数据，提高代码可读性。
利用 sqlite3.connect(detect_types=sqlite3.PARSE_DECLTYPES) 自动将日期时间、布尔值等转换为 Python 对象。
遵守最佳实践，如只选择必需的列，使用 with 语句管理连接和事务。
了解 SQLite ALTER TABLE 的局限性及其变通方案。
再次强调安全警告：避免在动态 DDL/DML 中直接拼接未经净化的用户输入。

通过这些全面而深入的知识，您将能够有效地从 SQLite 数据库中检索所需的数据，并构建功能强大、安全可靠的 Python 数据库应用程序。

清理测试文件 (为确保后续测试环境干净)

import os
files_to_clean = [
    'select_db.db',
    'comprehensive_select_db.db'
]
for f in files_to_clean:
    if os.path.exists(f):
        os.remove(f)
print("\n所有测试数据库文件已清理。")

亚马逊云科技技术品牌专区

更多推荐

ESP32-WROOM-32E-N4R2：自带2MB PSRAM的均衡之选，轻松应对复杂应用

亚马逊云科技技术品牌专区

JSM2318 40V N 沟道高速开关 MOSFET

随着消费电子、工业小型电源、物联网设备持续向小型化、高频化、高能效迭代，同步 Buck 降压电路对 MOS 管提出三重严苛要求：极低导通损耗、高速开关特性、优异抗干扰能力。市面上通用小信号 MOS 管往往存在栅电荷高、导通电阻偏大、高温性能衰减明显等问题，极易造成电源发热、转换效率下滑、EMI 干扰超标等工程难题。深耕功率半导体赛道的。

亚马逊云科技技术品牌专区

ESP32-WROOM-32E-H8：大容量Flash宽温模组，为工业物联网存储痛点而生

亚马逊云科技技术品牌专区

所有评论(0)

查看更多评论

weixin_42107409

@weixin_42107409

已为社区贡献114条内容

Python SQLite：从表中选择数据深度解析

weixin_42107409

目录

1. 引言：数据库的洞察之窗——数据查询

2. SQLite SELECT 语句基础

2.1 选择所有列 (SELECT *)

2.2 选择特定列 (SELECT column1, column2)

2.3 别名 (AS)

2.4 唯一值 (DISTINCT)

3. 筛选数据：WHERE 子句详解

3.1 比较运算符 (=, !=, <, >, <=, >=)

3.2 逻辑运算符 (AND, OR, NOT)

3.3 范围查询 (BETWEEN)

3.4 列表匹配 (IN, NOT IN)

3.5 模式匹配 (LIKE, GLOB)

3.5.1 LIKE 操作符 (%, _)

3.5.2 GLOB 操作符 (类似正则表达式)

3.6 空值判断 (IS NULL, IS NOT NULL)

4. 排序、限制与分页

4.1 排序结果 (ORDER BY)

4.1.1 升序 (ASC) 与降序 (DESC)

4.1.2 多列排序

4.2 限制结果数量 (LIMIT)

4.3 结果偏移 (OFFSET) 与分页

5. 在 Python sqlite3 中查询数据

5.1 建立连接与获取游标 (with 语句)

5.2 执行 SELECT 语句：参数绑定 (? 或 :name)

5.2.1 位置参数绑定

5.2.2 命名参数绑定

5.3 获取查询结果

5.3.1 cursor.fetchone()

5.3.2 cursor.fetchmany(size)

5.3.3 cursor.fetchall()

5.3.4 直接迭代游标

5.4 结果集增强：sqlite3.Row

5.5 自动类型转换 (detect_types) (续)

6. 高级查询概念 (简述)

6.1 聚合函数 (COUNT, SUM, AVG, MIN, MAX)

6.2 分组数据 (GROUP BY)

6.3 过滤分组 (HAVING)

6.4 表连接 (JOIN)

6.4.1 INNER JOIN

6.4.2 LEFT JOIN (或 LEFT OUTER JOIN)

6.4.3 SQLite 对 RIGHT JOIN 和 FULL JOIN 的限制

6.5 子查询 (Subqueries)

7. 错误处理

7.1 sqlite3.OperationalError

8. 最佳实践与注意事项

8.1 始终使用参数绑定进行查询 (防止 SQL 注入)

8.2 使用 with 语句管理数据库连接和游标

8.3 仅选择必需的列 (SELECT column1, column2 而非 SELECT *)

8.4 合理利用 LIMIT 和 OFFSET 进行分页

8.5 考虑索引以优化查询性能 (简要提及)

8.6 外键约束的启用 (PRAGMA foreign_keys = ON;)

8.7 理解并利用 sqlite3.Row 和 detect_types

8.8 ALTER TABLE 限制与常见变通方案

8.9 安全警告：避免动态 DDL/DML 中的用户输入

9. 综合代码示例

10. 总结

所有评论(0)

温馨提示：您尚未绑定手机号

weixin_42107409

2. SQLite `SELECT` 语句基础

2.1 选择所有列 (`SELECT *`)

2.2 选择特定列 (`SELECT column1, column2`)

2.3 别名 (`AS`)

2.4 唯一值 (`DISTINCT`)

3. 筛选数据：`WHERE` 子句详解

3.1 比较运算符 (`=`, `!=`, `<`, `>`, `<=`, `>=`)

3.2 逻辑运算符 (`AND`, `OR`, `NOT`)

3.3 范围查询 (`BETWEEN`)

3.4 列表匹配 (`IN`, `NOT IN`)

3.5 模式匹配 (`LIKE`, `GLOB`)

3.5.1 `LIKE` 操作符 (`%`, `_`)

3.5.2 `GLOB` 操作符 (类似正则表达式)

3.6 空值判断 (`IS NULL`, `IS NOT NULL`)

4.1 排序结果 (`ORDER BY`)

4.1.1 升序 (`ASC`) 与降序 (`DESC`)

4.2 限制结果数量 (`LIMIT`)

4.3 结果偏移 (`OFFSET`) 与分页

5. 在 Python `sqlite3` 中查询数据

5.1 建立连接与获取游标 (`with` 语句)

5.2 执行 `SELECT` 语句：参数绑定 (`?` 或 `:name`)

5.3.1 `cursor.fetchone()`

5.3.2 `cursor.fetchmany(size)`

5.3.3 `cursor.fetchall()`

5.4 结果集增强：`sqlite3.Row`

5.5 自动类型转换 (`detect_types`) (续)

6.1 聚合函数 (`COUNT`, `SUM`, `AVG`, `MIN`, `MAX`)

6.2 分组数据 (`GROUP BY`)

6.3 过滤分组 (`HAVING`)

6.4 表连接 (`JOIN`)

6.4.1 `INNER JOIN`

6.4.2 `LEFT JOIN` (或 `LEFT OUTER JOIN`)

6.4.3 SQLite 对 `RIGHT JOIN` 和 `FULL JOIN` 的限制

7.1 `sqlite3.OperationalError`

8.2 使用 `with` 语句管理数据库连接和游标

8.3 仅选择必需的列 (`SELECT column1, column2` 而非 `SELECT *`)

8.4 合理利用 `LIMIT` 和 `OFFSET` 进行分页

8.6 外键约束的启用 (`PRAGMA foreign_keys = ON;`)

8.7 理解并利用 `sqlite3.Row` 和 `detect_types`

8.8 `ALTER TABLE` 限制与常见变通方案