Python 操作 SQLite 教程

🐍 Python 操作 SQLite 教程 🐍

大家好！作为一名经验丰富的Python后端工程师和数据库专家，我将为您详细讲解如何使用 Python 内置的 `sqlite3` 模块来操作 SQLite 数据库。SQLite 因其轻量级、无服务器、零配置的特性，成为许多Python应用程序（如桌面应用、移动应用）以及开发测试场景的首选数据库。

核心思想： 使用 `sqlite3` 模块进行数据库操作的基本流程是：

连接数据库 (`connect()`)
创建游标 (`cursor()`)
执行 SQL 语句 (`execute()` 或 `executemany()`)
提交事务 (`commit()`)
关闭连接 (`close()`)

📦 一、`sqlite3` 模块简介

`sqlite3` 模块是 Python 标准库的一部分，无需额外安装。它提供了一个符合 DB-API 2.0 规范的接口，用于与 SQLite 数据库进行交互。

轻量级：不需要单独的数据库服务器进程。数据库就是一个文件。
零配置：无需复杂的安装或配置。
事务支持：完全支持 ACID 事务。
跨平台：数据库文件可以在不同操作系统之间直接复制使用。

🔗 二、连接到 SQLite 数据库

要连接到 SQLite 数据库，您需要调用 `sqlite3.connect()` 函数。如果指定的文件不存在，SQLite 会自动创建一个新的数据库文件。


import sqlite3
import os # 用于文件操作

# 定义数据库文件路径
DB_FILE = 'example.db'

# 如果数据库文件已存在，先删除，确保每次运行都是新的开始
if os.path.exists(DB_FILE):
    os.remove(DB_FILE)
    print(f"已删除现有数据库文件: {DB_FILE}")

# 连接到数据库。如果文件不存在，会创建一个新文件。
# :memory: 表示在内存中创建一个临时数据库，程序关闭后数据丢失。
try:
    conn = sqlite3.connect(DB_FILE)
    print(f"成功连接到数据库: {DB_FILE}")
except sqlite3.Error as e:
    print(f"数据库连接失败: {e}")
    conn = None # 确保连接对象为None，方便后续判断

# 关闭连接 (在实际应用中，通常在操作完成后关闭)
if conn:
    conn.close()
    print("数据库连接已关闭。")

📊 三、创建表

连接成功后，您需要创建一个游标 (Cursor) 对象来执行 SQL 语句。游标是您与数据库进行交互的接口。


import sqlite3
import os

DB_FILE = 'example.db'
if os.path.exists(DB_FILE):
    os.remove(DB_FILE)

try:
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor() # 获取游标对象
    print("数据库连接成功，游标已创建。")

    # 定义创建表的 SQL 语句
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        email TEXT UNIQUE
    );
    """
    # 执行 SQL 语句
    cursor.execute(create_table_sql)
    conn.commit() # 提交事务，使更改生效
    print("表 'users' 创建成功。")

except sqlite3.Error as e:
    print(f"数据库操作失败: {e}")
finally:
    if conn:
        conn.close()
        print("数据库连接已关闭。")

`IF NOT EXISTS` 的重要性： 在 `CREATE TABLE` 语句中使用 `IF NOT EXISTS` 可以防止在表已存在时再次创建表而引发错误。

➕ 四、插入数据 (CREATE)

插入数据可以使用 `INSERT INTO` 语句。建议使用参数化查询来防止 SQL 注入攻击。

1. 插入单条数据


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 插入单条数据
insert_sql = "INSERT INTO users (name, age, email) VALUES (?, ?, ?);"
user_data = ("Alice", 30, "alice@example.com")
cursor.execute(insert_sql, user_data)
conn.commit()
print(f"已插入用户: {user_data}")

conn.close()

2. 插入多条数据

使用 `executemany()` 方法可以高效地插入多条数据。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 准备多条数据
users_data = [
    ("Bob", 25, "bob@example.com"),
    ("Charlie", 35, "charlie@example.com"),
    ("David", 28, "david@example.com")
]

# 插入多条数据
insert_many_sql = "INSERT INTO users (name, age, email) VALUES (?, ?, ?);"
cursor.executemany(insert_many_sql, users_data)
conn.commit()
print(f"已批量插入 {len(users_data)} 条用户数据。")

conn.close()

🔍 五、查询数据 (READ)

查询数据使用 `SELECT` 语句。`fetch*` 方法用于获取查询结果。

1. 查询所有数据


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 查询所有用户
select_all_sql = "SELECT id, name, age, email FROM users;"
cursor.execute(select_all_sql)

# fetchall() 获取所有结果行
rows = cursor.fetchall()

print("\n--- 所有用户数据 ---")
for row in rows:
    print(row) # 默认返回元组

conn.close()

2. 查询特定条件的数据


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 查询年龄大于等于30的用户
select_filtered_sql = "SELECT name, age FROM users WHERE age >= ?;"
cursor.execute(select_filtered_sql, (30,)) # 注意参数是元组，即使只有一个参数

# fetchone() 获取下一行结果
print("\n--- 年龄 >= 30 的用户 (fetchone) ---")
row = cursor.fetchone()
if row:
    print(row)

# fetchmany(size) 获取指定数量的结果行
print("\n--- 再次获取，获取前2条 (fetchmany) ---")
cursor.execute(select_filtered_sql, (25,)) # 重新执行查询，因为fetchone/fetchall会消耗游标
rows_many = cursor.fetchmany(2)
for row in rows_many:
    print(row)

conn.close()

3. 以字典形式获取结果

默认情况下，`fetchall()` 和 `fetchone()` 返回的是元组。如果您想以字典（键值对）的形式访问结果，可以设置 `row_factory`。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
conn.row_factory = sqlite3.Row # 设置行工厂为 sqlite3.Row

cursor = conn.cursor()

select_all_sql = "SELECT id, name, age, email FROM users;"
cursor.execute(select_all_sql)

print("\n--- 所有用户数据 (字典形式) ---")
for row in cursor.fetchall():
    print(f"ID: {row['id']}, Name: {row['name']}, Age: {row['age']}, Email: {row['email']}")
    # 也可以像访问元组一样访问：print(row[0], row[1])

conn.close()

✏️ 六、更新数据 (UPDATE)

使用 `UPDATE` 语句来修改现有数据。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 更新 Alice 的年龄
update_sql = "UPDATE users SET age = ? WHERE name = ?;"
cursor.execute(update_sql, (31, "Alice"))
conn.commit()
print(f"\n已更新 Alice 的年龄。影响行数: {cursor.rowcount}")

conn.close()

🗑️ 七、删除数据 (DELETE)

使用 `DELETE FROM` 语句来删除数据。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 删除年龄大于30的用户
delete_sql = "DELETE FROM users WHERE age > ?;"
cursor.execute(delete_sql, (30,))
conn.commit()
print(f"\n已删除年龄大于30的用户。影响行数: {cursor.rowcount}")

conn.close()

🔄 八、事务管理

`sqlite3` 模块默认开启事务。每次执行 `execute()` 或 `executemany()` 后，更改会暂存，直到调用 `commit()` 才会持久化到数据库文件。如果发生错误或您想撤销更改，可以调用 `rollback()`。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

try:
    # 尝试插入一个重复的 email，会触发 UNIQUE 约束错误
    cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?);", ("Frank", 22, "bob@example.com"))
    conn.commit() # 如果没有错误，提交
    print("成功插入 Frank。")
except sqlite3.IntegrityError as e:
    print(f"\n插入 Frank 失败（事务回滚）: {e}")
    conn.rollback() # 回滚事务，撤销之前的操作

# 验证 Bob@example.com 是否仍然存在 (因为 Frank 的插入失败了)
cursor.execute("SELECT name, email FROM users WHERE email = ?;", ("bob@example.com",))
print(f"当前 bob@example.com 的用户: {cursor.fetchone()}")

conn.close()

上下文管理器： 推荐使用 `with` 语句来自动管理数据库连接和事务，确保连接被正确关闭，并且在发生异常时自动回滚。


import sqlite3

DB_FILE = 'example.db'

try:
    with sqlite3.connect(DB_FILE) as conn: # conn 是连接对象
        cursor = conn.cursor()

        # 正常操作，如果无异常，conn.commit() 会自动执行
        cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?);", ("Grace", 27, "grace@example.com"))
        print("Grace 已插入。")

        # 模拟一个会失败的操作，导致自动回滚
        print("尝试插入一个重复的 email (预期失败)...")
        cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?);", ("Henry", 29, "alice@example.com"))
        print("Henry 已插入。") # 这行不会被执行到
except sqlite3.IntegrityError as e:
    print(f"发生错误，事务自动回滚: {e}")
    # conn.rollback() 会自动处理，这里不需要显式调用
except sqlite3.Error as e:
    print(f"其他数据库错误: {e}")
finally:
    # 检查 Grace 是否被插入 (如果没有发生错误)
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT name, email FROM users WHERE name = 'Grace';")
        print(f"查询 Grace: {cursor.fetchone()}") # 如果没有其他错误，Grace 应该存在
        cursor.execute("SELECT name, email FROM users WHERE name = 'Henry';")
        print(f"查询 Henry: {cursor.fetchone()}") # Henry 不应该存在

❌ 九、错误处理

`sqlite3` 模块会抛出 `sqlite3.Error` 及其子类异常（如 `IntegrityError`、`OperationalError` 等）。捕获这些异常是健壮程序的重要组成部分。


import sqlite3

DB_FILE = 'example.db'

try:
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()

    # 尝试创建一个已存在的表 (如果上面没有删除DB_FILE)
    cursor.execute("CREATE TABLE users (id INTEGER);")
    conn.commit()
except sqlite3.OperationalError as e:
    print(f"\n操作错误 (表已存在): {e}")
except sqlite3.Error as e:
    print(f"\n通用数据库错误: {e}")
finally:
    if conn:
        conn.close()

🚀 十、高级主题与最佳实践

1. 索引 (Indexes)

为经常用于查询条件的列创建索引可以显著提高查询速度。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 为 name 列创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_name ON users (name);")
conn.commit()
print("\n索引 'idx_users_name' 已创建。")

conn.close()

2. 数据库文件大小优化

删除数据后，数据库文件大小可能不会立即减小。可以使用 `VACUUM` 命令来回收空间。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 执行 VACUUM 操作
cursor.execute("VACUUM;")
conn.commit()
print("\n数据库已执行 VACUUM 操作，回收空间。")

conn.close()

3. PRAGMA 指令

SQLite 提供 `PRAGMA` 指令来控制数据库的行为。

`PRAGMA foreign_keys = ON;`：启用外键约束检查（默认是关闭的）。
`PRAGMA journal_mode = WAL;`：使用 WAL (Write-Ahead Logging) 日志模式，可以提高并发性能。


import sqlite3

DB_FILE = 'example.db'

conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()

# 启用外键约束
cursor.execute("PRAGMA foreign_keys = ON;")
# 设置 WAL 模式
cursor.execute("PRAGMA journal_mode = WAL;")

conn.close()
print("\n外键约束和 WAL 日志模式已设置。")

4. 连接图示

理解 Python 应用程序与 SQLite 数据库文件的交互流程：

graph TD A[Python 应用代码] --> B[sqlite3.connect（）]; B --> C{数据库文件: .db}; C --> D[Connection 对象]; D --> E[Cursor 对象]; E -- SQL 查询/命令 --> C; E -- 结果集/影响行数 --> D; D -- commit（） --> C; D -- rollback（） --> C; D -- close（） --> C;

🎉 结语

通过本教程，您应该已经全面掌握了使用 Python `sqlite3` 模块操作 SQLite 数据库的各项技能，从基本的 CRUD 操作到事务管理、错误处理和一些高级优化技巧。SQLite 因其简单、高效的特点，在许多场景下都是一个非常实用的选择。希望这份教程对您有所帮助，祝您在 Python 数据库编程的道路上一切顺利！

目录导航

Python 操作 SQLite 教程

预览结果：