sqlite3模块 — SQLite数据库操作

Python标准库精讲专题 · 数据持久化篇 · 掌握SQLite数据库操作

专题：Python标准库精讲系统学习

关键词：Python, 标准库, sqlite3, SQLite, 数据库, 连接, 游标, 事务, execute, 参数化查询, row_factory

一、sqlite3模块概述

sqlite3是Python标准库中内置的SQLite数据库接口模块，遵循Python DB-API 2.0规范（PEP 249）。SQLite本身是一个轻量级、无服务器、零配置、事务性SQL数据库引擎，其核心库以C语言编写，广泛应用于嵌入式设备、移动应用、桌面软件及中小型Web应用的数据存储场景。

作为Python标准库的一部分，sqlite3无需任何外部依赖即可使用，是Python生态中最便捷、最稳定的数据持久化方案之一。它支持完整的SQL语法（包括事务、触发器、视图、索引），能够将整个数据库存储为单个跨平台文件，非常适合作为应用程序的原生存储引擎。

sqlite3模块的核心设计理念是简洁与易用：通过Connection对象管理数据库连接，通过Cursor对象执行SQL语句并获取结果集。对于绝大多数单进程、中小规模的数据存储需求，sqlite3提供了远超文件存储的查询能力和数据完整性保障，同时又避免了传统客户端-服务器数据库（如MySQL、PostgreSQL）的部署和运维复杂度。

核心特性一览：

无服务器架构 — 数据库引擎直接集成在应用中，无需单独安装和配置数据库服务

零配置 — 创建数据库即创建文件或指定内存模式，无需管理员权限或配置文件

ACID事务 — 支持原子性、一致性、隔离性、持久性，确保数据安全

跨平台 — 数据库文件在Windows、macOS、Linux之间完全兼容

Python原生 — 标准库内置，import sqlite3即可使用，无第三方依赖

二、Connection连接 — connect创建/内存数据库:memory:/uri参数

sqlite3模块的核心入口是sqlite3.connect()函数，它接受一个数据库路径参数并返回Connection对象。Connection对象代表了与SQLite数据库文件的通信通道，所有数据库操作都基于它展开。

基本文件数据库连接

import sqlite3 # 创建或打开一个SQLite数据库文件 conn = sqlite3.connect('example.db') # 操作完成后关闭连接 conn.close()

当指定的数据库文件不存在时，sqlite3会自动创建该文件。这一点与传统数据库管理系统不同——SQLite采取"文件即数据库"的策略，数据库文件后缀名可以是任意名称（通常使用.db、.sqlite或.sqlite3）。

内存数据库 :memory:

SQLite支持完全在内存中运行的数据库模式，适用于临时数据处理、单元测试和性能敏感场景。使用特殊字符串":memory:"作为数据库路径即可创建内存数据库：

import sqlite3 # 创建内存数据库 conn = sqlite3.connect(':memory:') # 内存数据库运行在RAM中，速度极快 # 但连接关闭后所有数据自动销毁 conn.execute('''CREATE TABLE test (id INT, val TEXT)''') conn.execute('''INSERT INTO test VALUES (?, ?)''', (1, 'hello')) result = conn.execute('''SELECT * FROM test''').fetchall() print(result) # [(1, 'hello')] conn.close()

内存数据库在以下场景中尤为有用：

URI模式连接

从Python 3.4开始，sqlite3支持以URI形式传递连接参数，需要通过uri=True参数启用。URI模式允许在连接字符串中指定多个配置选项：

import sqlite3 # 使用URI模式创建内存数据库 conn = sqlite3.connect('file::memory:?cache=shared', uri=True) # 以只读方式打开现有数据库 conn = sqlite3.connect('file:example.db?mode=ro', uri=True) # 设置数据库缓存大小为10000页 conn = sqlite3.connect('file:example.db?cache=private', uri=True) conn.close()

URI模式支持的主要参数包括：mode=ro（只读）、mode=rw（读写，文件必须存在）、mode=rwc（读写，不存在则创建）、cache=shared（共享缓存）、cache=private（私有缓存）。合理利用URI模式可以更精细地控制数据库行为。

Connection对象的核心方法

方法说明
conn.cursor()创建游标对象
conn.commit()提交当前事务
conn.rollback()回滚当前事务
conn.close()关闭数据库连接
conn.execute(sql, params)快捷执行SQL（自动创建并返回游标）
conn.executemany(sql, seq)批量执行相同SQL
conn.executescript(sql_script)执行多条SQL语句（以;分隔）
conn.set_trace_callback(cb)设置SQL跟踪回调函数
conn.row_factory设置行工厂类/函数
conn.text_factory控制文本编码转换行为
conn.total_changes返回连接以来修改的行数

最佳实践：使用with语句管理数据库连接可以自动提交或回滚事务，并确保连接被正确关闭。这是Python中管理sqlite3连接的推荐方式。

import sqlite3 # 使用上下文管理器自动管理连接 with sqlite3.connect('example.db') as conn: conn.execute('''CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER )''') conn.execute('''INSERT INTO users (name, age) VALUES (?, ?)''', ('Alice', 30)) # 退出with块时自动提交，即使发生异常也会自动回滚 # 退出with块后连接自动关闭

三、Cursor游标 — execute/executemany/executescript、fetchone/fetchmany/fetchall

游标（Cursor）是Python DB-API规范中定义的核心对象，负责执行SQL语句并管理结果集。在sqlite3中，游标通过Connection对象的cursor()方法创建。

游标的创建与基本使用

import sqlite3 with sqlite3.connect('example.db') as conn: cur = conn.cursor() # 创建表 cur.execute('''CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT, price REAL, stock INTEGER )''') # 插入单条记录 cur.execute('''INSERT INTO products VALUES (1, '键盘', 99.9, 50)''') conn.commit()

execute — 执行单条SQL语句

execute()用于执行单条SQL语句，是最常用的方法。它支持参数化查询（详见第四章），可以有效防止SQL注入。execute()执行后，如果语句是查询，结果集将缓存在游标中供后续fetch操作读取。

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE items (id INT, name TEXT)''') # 插入数据（使用参数化查询） cur.execute('''INSERT INTO items VALUES (?, ?)''', (1, 'item1')) cur.execute('''INSERT INTO items VALUES (?, ?)''', (2, 'item2')) # 查询数据 cur.execute('''SELECT * FROM items WHERE id = ?''', (1,)) row = cur.fetchone() print(row) # (1, 'item1')

executemany — 批量执行

executemany()接受一个SQL模板和一个可迭代的参数序列，为每个参数执行一次SQL语句。相比循环调用execute()，executemany()的性能显著更优，因为它将多次SQL执行合并为一次底层调用。

import sqlite3 import csv with sqlite3.connect('example.db') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE IF NOT EXISTS employees ( id INT, name TEXT, department TEXT )''') # 批量插入数据 employees_data = [ (1, '张三', '技术部'), (2, '李四', '市场部'), (3, '王五', '技术部'), (4, '赵六', '财务部'), ] cur.executemany('''INSERT INTO employees VALUES (?, ?, ?)''', employees_data) conn.commit() # 验证结果 cur.execute('''SELECT COUNT(*) FROM employees''') print(cur.fetchone()[0]) # 4

executescript — 执行SQL脚本

executescript()用于执行包含多条SQL语句的脚本字符串，语句之间使用分号（;）分隔。这是一个便捷的批量初始化工具，尤其适用于执行数据库初始化脚本或迁移脚本。注意：executescript()会首先提交当前事务中的任何未完成更改。

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() # 执行SQL脚本——多条语句一次性执行 schema = ''' CREATE TABLE customers ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE ); CREATE TABLE orders ( id INTEGER PRIMARY KEY, customer_id INTEGER REFERENCES customers(id), amount REAL, order_date TEXT ); INSERT INTO customers VALUES (1, 'Alice', 'alice@example.com'); INSERT INTO customers VALUES (2, 'Bob', 'bob@example.com'); ''' # executescript会先提交当前事务 cur.executescript(schema) # 验证 cur.execute('''SELECT COUNT(*) FROM customers''') print(cur.fetchone()[0]) # 2

fetchone / fetchmany / fetchall — 结果集获取

execute()执行查询后，通过以下三种方法获取结果集：

方法返回说明
fetchone()单行元组或None获取下一行结果，无数据时返回None
fetchmany(size)行元组列表最多获取size行结果，默认值为游标的arraysize属性
fetchall()行元组列表获取所有剩余结果行
import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE numbers (val INT)''') cur.executemany('''INSERT INTO numbers VALUES (?)''', [(i,) for i in range(100)]) conn.commit() # --- fetchone --- cur.execute('''SELECT val FROM numbers ORDER BY val''') first = cur.fetchone() print(first) # (0,) second = cur.fetchone() print(second) # (1,) # --- fetchmany --- batch = cur.fetchmany(5) print(batch) # [(2,), (3,), (4,), (5,), (6,)] # --- fetchall --- remaining = cur.fetchall() print(len(remaining)) # 93 # 游标耗尽后fetchone返回None print(cur.fetchone()) # None

游标作为迭代器

Cursor对象实现了迭代器协议，可以直接用于for循环，避免显式调用fetch方法。这种方式在处理大量数据时更加简洁和内存高效：

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE scores (name TEXT, score INT)''') cur.executemany('''INSERT INTO scores VALUES (?, ?)''', [('Alice', 95), ('Bob', 87), ('Charlie', 92)]) conn.commit() cur.execute('''SELECT * FROM scores ORDER BY score DESC''') # 直接迭代游标 for row in cur: print(f'{row[0]}: {row[1]}') # 输出: # Alice: 95 # Charlie: 92 # Bob: 87

四、参数化查询 — ?占位符vs命名占位符(:name)、SQL注入防护

参数化查询是sqlite3最重要的安全特性之一。它通过占位符将SQL语句结构与数据分离，从根本上杜绝了SQL注入攻击的风险，同时还能提升重复查询的性能（SQLite可以缓存和复用查询计划）。

? 占位符风格（位置参数）

使用问号?作为占位符，参数以元组形式传递。这是DB-API 2.0规范定义的qmark风格，也是最常用的一种：

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE users (id INT, name TEXT, age INT)''') # ? 占位符，参数为元组 cur.execute('''INSERT INTO users VALUES (?, ?, ?)''', (1, 'Alice', 30)) # 多个占位符匹配多个参数 cur.execute('''SELECT * FROM users WHERE name = ? AND age = ?''', ('Alice', 30)) print(cur.fetchone()) # (1, 'Alice', 30) # 参数个数必须与?数量完全一致 # cur.execute('INSERT INTO users VALUES (?, ?)', (1,)) # 错误！

:name 命名占位符风格

使用:name形式的命名占位符，参数以字典形式传递。命名占位符在参数较多或同一参数多次出现时更具可读性和可维护性：

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE users (id INT, name TEXT, age INT)''') # 命名占位符，参数为字典 cur.execute('''INSERT INTO users VALUES (:id, :name, :age)''', {'id': 1, 'name': 'Bob', 'age': 25}) # 同一参数可在SQL中多次出现 cur.execute(''' SELECT * FROM users WHERE age >= :min_age AND age <= :max_age ''', {'min_age': 20, 'max_age': 30}) # 命名占位符也兼容序列（按位置匹配） cur.execute('''SELECT * FROM users WHERE name = :name''', {'name': 'Bob'}) print(cur.fetchone()) # (1, 'Bob', 25)

SQL注入防护（重点强调）

SQL注入是最严重的Web安全漏洞之一，其成因是将用户输入直接拼接到SQL语句中。使用字符串格式化拼接SQL是绝对禁止的操作：

# ============== 危险！切勿使用 ============== # 以下方式存在严重的SQL注入风险 user_input = "' OR '1'='1" # 恶意输入 # 危险：直接拼接用户输入 query = f"SELECT * FROM users WHERE name = '{user_input}'" # 实际执行: SELECT * FROM users WHERE name = '' OR '1'='1' # 会返回所有用户数据，造成严重信息泄露 # 更危险：DROP TABLE 等破坏性操作 evil_input = "'; DROP TABLE users; --" query = f"DELETE FROM users WHERE name = '{evil_input}'" # 实际执行: DELETE FROM users WHERE name = ''; DROP TABLE users; --' # 整个users表会被删除！！！ # ============================================

核心原则：永远不要使用字符串格式化（%、f-string、+）拼接SQL语句中的参数值。

始终使用参数化查询（?占位符或:name命名占位符），由数据库驱动负责正确的转义和引用。

正确的做法是使用参数化查询，将用户输入作为参数传入：

import sqlite3 # ============== 安全方式 ============== # 使用参数化查询，数据库驱动自动处理转义 user_input = "' OR '1'='1" with sqlite3.connect('example.db') as conn: cur = conn.cursor() # 安全：使用?占位符 cur.execute('''SELECT * FROM users WHERE name = ?''', (user_input,)) # 实际执行: SELECT * FROM users WHERE name = ''' OR ''1''=''1' # 用户输入被正确转义为字符串字面量，不会改变查询逻辑 # 安全：使用:name命名占位符 cur.execute('''SELECT * FROM users WHERE name = :name''', {'name': user_input}) # 效果与?占位符相同，安全可靠 # ====================================

参数化查询的两个关键优势：

五、事务管理 — 隐式事务/显式commit/rollback、isolation_level

事务是数据库保证数据一致性的核心机制。SQLite默认每条SQL语句自动开启和提交事务（自动提交模式），但sqlite3模块默认使用隐式事务管理，要求开发者显式调用commit()或rollback()来结束事务。

隐式事务与显式提交

在默认配置下，执行数据修改语句（INSERT、UPDATE、DELETE）时，sqlite3会自动开始一个新事务，但不会自动提交。开发者必须调用conn.commit()使更改持久化，或调用conn.rollback()撤销更改：

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE accounts (id INT, name TEXT, balance REAL)''') # 插入数据——隐式开启事务 cur.execute('''INSERT INTO accounts VALUES (1, 'Alice', 1000.0)''') cur.execute('''INSERT INTO accounts VALUES (2, 'Bob', 500.0)''') # 此时尚未持久化，其他连接看不到这些数据 # 必须提交才能生效 conn.commit() # 如果此时发生错误或需要撤销 cur.execute('''UPDATE accounts SET balance = balance - 200 WHERE id = 1''') # 发现不应该扣款，回滚 conn.rollback() # balance恢复为1000 # 验证回滚结果 cur.execute('''SELECT balance FROM accounts WHERE id = 1''') print(cur.fetchone()[0]) # 1000.0

使用上下文管理器自动管理事务

Connection对象可以作为上下文管理器使用。进入with块时不会自动开启事务，但离开with块时，如果没有发生异常则自动commit()，如果发生异常则自动rollback()。这是一种简洁且安全的事务管理模式：

import sqlite3 def transfer_funds(from_id, to_id, amount): """安全的资金转账操作""" with sqlite3.connect('bank.db') as conn: cur = conn.cursor() # 扣减转出账户 cur.execute('''UPDATE accounts SET balance = balance - ? WHERE id = ?''', (amount, from_id)) # 增加转入账户 cur.execute('''UPDATE accounts SET balance = balance + ? WHERE id = ?''', (amount, to_id)) # 如果任何一步失败，with块退出时会自动rollback() # 确保转账的原子性：要么全部成功，要么全部失败 # 无异常时自动commit() # 使用示例 transfer_funds(1, 2, 100.0) # 如果transfer_funds中途崩溃，不会出现钱被扣减但未到账的情况

isolation_level 隔离级别控制

sqlite3通过isolation_level属性控制事务行为。该属性在创建连接时通过connect()的isolation_level参数设置：

isolation_level值行为
"" 空字符串（默认）自动提交模式关闭，需要显式调用commit/rollback
"DEFERRED"（延迟）首次访问数据库时才获取锁，默认的显式事务模式
"IMMEDIATE"（立即）立即获取写锁，防止其他连接并发写入
"EXCLUSIVE"（排他）获取排他锁，防止任何其他连接读写
None启用自动提交模式，每条SQL语句自成一事务
import sqlite3 # 自动提交模式：每条语句自动提交，无需显式commit conn = sqlite3.connect(':memory:', isolation_level=None) cur = conn.cursor() cur.execute('''CREATE TABLE test (id INT)''') cur.execute('''INSERT INTO test VALUES (1)''') # 自动提交 # 无需 conn.commit()，数据已持久化 conn.close() # 显式事务模式（默认）：需要手动commit conn = sqlite3.connect(':memory:') # isolation_level="" cur = conn.cursor() cur.execute('''CREATE TABLE test (id INT)''') cur.execute('''INSERT INTO test VALUES (1)''') conn.commit() # 必须显式提交 conn.close() # 指定IMMEDIATE事务模式 conn = sqlite3.connect(':memory:', isolation_level='IMMEDIATE') # 执行修改语句时立即获取写锁 cur = conn.cursor() cur.execute('''CREATE TABLE test (id INT)''') cur.execute('''INSERT INTO test VALUES (1)''') conn.commit() conn.close()

注意：当isolation_level为None（自动提交模式）时，每次调用execute()、executemany()都会立即提交当前事务。这意味着无法回滚已经执行的操作，因此在需要原子性的场景中应使用默认模式或显式事务。

保存点（SAVEPOINT）

除了完整的事务提交和回滚，SQLite还支持保存点机制，允许在事务内部设置回滚点，实现细粒度的部分回滚：

import sqlite3 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE items (id INT, val TEXT)''') cur.execute('''INSERT INTO items VALUES (1, 'a')''') cur.execute('''INSERT INTO items VALUES (2, 'b')''') # 设置保存点 cur.execute('''SAVEPOINT sp1''') cur.execute('''INSERT INTO items VALUES (3, 'c')''') # 回滚到保存点，撤销保存点后的更改 cur.execute('''ROLLBACK TO sp1''') conn.commit() cur.execute('''SELECT COUNT(*) FROM items''') print(cur.fetchone()[0]) # 2（id=3被回滚了）

六、行工厂Row — row_factory=Row属性访问、自定义行工厂、字典行工厂

默认情况下，sqlite3的查询结果以元组（tuple）形式返回，只能通过索引访问列值。这在列数较多或列顺序不确定时很不方便。行工厂（Row Factory）机制允许自定义结果行的返回格式，使数据访问更加直观和灵活。

sqlite3.Row 行工厂

sqlite3.Row是一个内置的行包装类，同时支持索引访问和列名访问，还支持迭代、相等性测试和len()操作：

import sqlite3 with sqlite3.connect(':memory:') as conn: # 启用Row行工厂 conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute('''CREATE TABLE users ( id INT, name TEXT, age INT, email TEXT )''') cur.execute('''INSERT INTO users VALUES (1, 'Alice', 30, 'alice@example.com')''') conn.commit() cur.execute('''SELECT * FROM users''') row = cur.fetchone() # 通过列名访问——清晰直观 print(row['name']) # Alice print(row['age']) # 30 print(row['email']) # alice@example.com # 通过索引访问——保持兼容 print(row[0]) # 1 (id) print(row[1]) # Alice (name) # 支持len()和keys() print(len(row)) # 4 print(row.keys()) # ['id', 'name', 'age', 'email'] # 支持解包 id_, name, age, email = row print(name) # Alice # Row对象可转换为字典 row_dict = dict(row) print(row_dict) # {'id': 1, 'name': 'Alice', 'age': 30, 'email': 'alice@example.com'}

自定义行工厂

通过设置row_factory属性为自定义函数，可以完全控制结果行的格式。行工厂函数接收两个参数：游标对象和原始行元组，返回任意格式的结果：

import sqlite3 from dataclasses import dataclass # 自定义数据类行工厂 @dataclass class User: id: int name: str age: int def user_row_factory(cursor, row): """将数据库行转换为User数据类实例""" columns = [desc[0] for desc in cursor.description] data = dict(zip(columns, row)) return User(**data) # 也可以使用命名元组 from collections import namedtuple def namedtuple_row_factory(cursor, row): """将数据库行转换为命名元组""" columns = [desc[0] for desc in cursor.description] NT = namedtuple('Row', columns) return NT(*row) # 使用字典作为行工厂 def dict_row_factory(cursor, row): """将数据库行转换为字典""" columns = [desc[0] for desc in cursor.description] return dict(zip(columns, row)) # 测试 with sqlite3.connect(':memory:') as conn: conn.row_factory = dict_row_factory # 使用字典行工厂 cur = conn.cursor() cur.execute('''CREATE TABLE users (id INT, name TEXT, age INT)''') cur.execute('''INSERT INTO users VALUES (1, 'Alice', 30)''') conn.commit() cur.execute('''SELECT * FROM users''') row = cur.fetchone() print(row) # {'id': 1, 'name': 'Alice', 'age': 30} print(row['name']) # Alice（字典访问）

字典行工厂的便捷实现

从Python 3.12开始，sqlite3模块新增sqlite3.row_factories命名空间，但在此之前，最通用的字典行工厂实现就是上面的dict_row_factory函数。许多开发者将此函数定义为工具函数，在项目各处复用：

import sqlite3 def dict_factory(cursor, row): """将查询行转换为字典的通用行工厂""" fields = [column[0] for column in cursor.description] return {key: value for key, value in zip(fields, row)} # 使用示例 with sqlite3.connect('example.db') as conn: conn.row_factory = dict_factory cur = conn.cursor() cur.execute('''SELECT id, name, email FROM users''') users = cur.fetchall() for user in users: print(user['name'], user['email']) # 每行都是字典，访问方式统一且安全

行工厂选择建议：

大部分场景推荐使用sqlite3.Row——它兼顾了性能（使用C语言实现）和便利性（同时支持索引和列名访问）。

需要将行转换为JSON序列化对象时，使用字典行工厂。

需要类型注解和IDE智能提示时，使用数据类或Pydantic模型行工厂。

七、适配器与转换器 — register_adapter/register_converter自定义类型映射、datetime等内置适配

SQLite本身仅支持五种基本数据类型：NULL、INTEGER、REAL、TEXT、BLOB。但Python具有更丰富的类型系统（如datetime、decimal、UUID等）。适配器（Adapter）和转换器（Converter）机制实现了Python类型与SQLite类型之间的双向映射。

适配器（Adapter）— Python类型 → SQLite存储

sqlite3.register_adapter(type, callable)注册将Python类型转换为SQLite可存储类型的函数。该函数接收Python对象，返回SQLite可接受的类型（int、float、str、bytes、None）：

import sqlite3 from datetime import datetime import json # 为datetime类型注册适配器 # datetime → 字符串（ISO格式） sqlite3.register_adapter(datetime, lambda dt: dt.isoformat()) # 为字典类型注册适配器 # dict → JSON字符串 sqlite3.register_adapter(dict, lambda d: json.dumps(d, ensure_ascii=False)) # 为列表类型注册适配器 # list → JSON字符串 sqlite3.register_adapter(list, lambda l: json.dumps(l, ensure_ascii=False)) # 使用示例 with sqlite3.connect(':memory:') as conn: cur = conn.cursor() cur.execute('''CREATE TABLE records ( id INT, created_at TEXT, metadata TEXT, tags TEXT )''') now = datetime.now() meta = {'source': 'API', 'version': 2} tag_list = ['python', 'sqlite3', 'adapter'] # datetime、dict、list被自动转换为字符串存储 cur.execute('''INSERT INTO records VALUES (?, ?, ?, ?)''', (1, now, meta, tag_list)) conn.commit() # 查看实际存储的内容（字符串形式） cur.execute('''SELECT * FROM records''') row = cur.fetchone() print(row) # 输出: (1, '2026-05-05T23:57:03', '{"source": "API", "version": 2}', # '["python", "sqlite3", "adapter"]')

转换器（Converter）— SQLite存储 → Python类型

sqlite3.register_converter(typename, callable)注册将从SQLite读取的字符串（或BLOB）转换回Python类型的函数。转换器通过SQL类型名称与数据库中的列类型进行匹配，因此在创建连接时必须指定detect_types=sqlite3.PARSE_DECLTYPES：

import sqlite3 from datetime import datetime import json # 注册适配器（Python→SQLite） sqlite3.register_adapter(datetime, lambda dt: dt.isoformat()) # 注册转换器（SQLite→Python） # 注意：此处的类型名必须与SQL建表语句中的类型名一致 sqlite3.register_converter('datetime', lambda s: datetime.fromisoformat(s.decode('utf-8'))) # 创建连接时启用类型检测 conn = sqlite3.connect( ':memory:', detect_types=sqlite3.PARSE_DECLTYPES # PARSE_DECLTYPES: 根据SQL声明类型匹配转换器 ) cur = conn.cursor() # 建表时列类型使用 'datetime'（与注册的转换器名称对应） cur.execute('''CREATE TABLE events ( id INT, event_time datetime, description TEXT )''') now = datetime.now() cur.execute('''INSERT INTO events VALUES (?, ?, ?)''', (1, now, '测试事件')) conn.commit() cur.execute('''SELECT * FROM events''') row = cur.fetchone() print(type(row[1])) # print(row[1]) # 2026-05-05 23:57:03（Python datetime对象） print(row[1].year) # 2026（可以正常调用datetime的方法）

内置适配器与PARSE_DECLTYPES / PARSE_COLNAMES

sqlite3为datetime.date、datetime.datetime、datetime.time、datetime.timedelta等类型提供了内置适配器。只需在连接时启用detect_types参数即可自动生效：

import sqlite3 from datetime import date, datetime, timedelta # 启用声明类型检测和列名检测 conn = sqlite3.connect( ':memory:', detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES ) cur = conn.cursor() cur.execute('''CREATE TABLE events ( id INT, event_date DATE, event_time TIMESTAMP, duration INTERVAL )''') today = date.today() now = datetime.now() delta = timedelta(hours=2, minutes=30) cur.execute('''INSERT INTO events VALUES (?, ?, ?, ?)''', (1, today, now, delta)) conn.commit() cur.execute('''SELECT * FROM events''') row = cur.fetchone() print(type(row[1])) # print(type(row[2])) # print(type(row[3])) # # 全部自动转换为对应的Python类型！
SQLite类型名Python类型内置适配支持
DATEdatetime.date内置
TIMESTAMPdatetime.datetime内置
TIMEdatetime.time内置
INTERVALdatetime.timedelta内置

detect_types支持两种检测模式，可以组合使用：

# PARSE_COLNAMES 示例 import sqlite3 conn = sqlite3.connect( ':memory:', detect_types=sqlite3.PARSE_COLNAMES ) cur = conn.cursor() cur.execute('''CREATE TABLE items (id INT, ts TEXT)''') cur.execute('''INSERT INTO items VALUES (1, '2026-05-05 12:00:00')''') conn.commit() # 使用列名标注指定转换类型 cur.execute(''' SELECT id, ts AS "ts [TIMESTAMP]" FROM items ''') row = cur.fetchone() print(type(row[1])) # （即使原始存储为TEXT） # [TIMESTAMP]标注告诉sqlite3使用TIMESTAMP转换器

自定义类型的完整适配示例

import sqlite3 from decimal import Decimal import json # 为Decimal类型注册适配器和转换器 sqlite3.register_adapter(Decimal, lambda d: str(d)) # Decimal → 字符串 sqlite3.register_converter('decimal', lambda s: Decimal(s.decode('utf-8'))) # 为UUID类型注册适配器和转换器 import uuid sqlite3.register_adapter(uuid.UUID, lambda u: str(u)) # UUID → 字符串 sqlite3.register_converter('uuid', lambda s: uuid.UUID(s.decode('utf-8'))) # 使用自定义类型 conn = sqlite3.connect( ':memory:', detect_types=sqlite3.PARSE_DECLTYPES ) cur = conn.cursor() cur.execute('''CREATE TABLE products ( id INT, price decimal, product_id uuid )''') price = Decimal('99.99') pid = uuid.uuid4() cur.execute('''INSERT INTO products VALUES (?, ?, ?)''', (1, price, pid)) conn.commit() cur.execute('''SELECT * FROM products''') row = cur.fetchone() print(type(row[1])) # print(type(row[2])) # print(row) # (1, Decimal('99.99'), UUID('...')) # 完美还原Python原生类型！

八、核心总结

sqlite3模块核心要点速览：

1. Connection是数据库操作的总入口，通过sqlite3.connect()创建

2. Cursor负责执行SQL和获取结果，通过conn.cursor()获取

3. 参数化查询使用?或:name占位符，严禁字符串拼接

4. 事务管理默认需要显式commit/rollback，推荐使用with语句自动管理

5. 行工厂（row_factory）让列访问更直观——Row、字典、数据类任选

6. 适配器/转换器实现Python类型与SQLite类型的双向映射

设计原则

常见陷阱与解决方案

陷阱表现解决方案
数据库被锁定sqlite3.OperationalError: database is locked使用连接池或设置timeout参数；确保及时关闭连接；避免跨线程共享连接
忘记提交事务数据写入后其他连接看不到更改使用with conn:语句自动提交；或在每次修改后立即commit()
SQL注入风险用户输入破坏查询逻辑始终使用?或:name参数化查询，严禁f-string/%格式化拼接参数
类型丢失datetime等类型读出后变成字符串注册适配器/转换器，使用detect_types参数创建连接
跨线程使用异常或数据错乱每个线程创建独立连接；或使用check_same_thread=False（谨慎）
并发写入冲突多进程同时写入同一数据库SQLite适合单进程应用；高并发写入场景应考虑PostgreSQL等

最佳实践清单

速查备忘录

import sqlite3 # ─── 连接数据库 ─── conn = sqlite3.connect('data.db') # 文件数据库 conn = sqlite3.connect(':memory:') # 内存数据库 conn = sqlite3.connect('file:data.db?mode=ro', uri=True) # 只读模式 # ─── 事务设置 ─── conn.isolation_level = None # 自动提交模式 conn.execute('PRAGMA journal_mode=WAL') # 启用WAL模式 conn.execute('PRAGMA foreign_keys=ON') # 启用外键约束 # ─── 行工厂 ─── conn.row_factory = sqlite3.Row # 支持列名+索引访问 conn.row_factory = dict_factory # 自定义字典工厂 # ─── 类型检测 ─── conn = sqlite3.connect('data.db', detect_types=sqlite3.PARSE_DECLTYPES) # 启用类型自动转换 # ─── 参数化查询 ─── cur.execute('SELECT * FROM t WHERE id = ?', (val,)) # ? 占位符 cur.execute('SELECT * FROM t WHERE id = :id', {'id': val}) # :name 占位符 # ─── 结果获取 ─── row = cur.fetchone() # 单行 / None rows = cur.fetchmany(10) # 最多10行 rows = cur.fetchall() # 全部剩余行 for row in cur: ... # 迭代器方式 # ─── 事务控制 ─── conn.commit() # 提交事务 conn.rollback() # 回滚事务 with conn: ... # 自动提交/回滚 # ─── 批量操作 ─── cur.executemany('INSERT INTO t VALUES (?, ?)', data_seq) # 批量插入 cur.executescript(multi_sql_string) # 批量执行脚本 # ─── 适配器与转换器 ─── sqlite3.register_adapter(MyType, to_sql_func) sqlite3.register_converter('mytype', from_sql_func)

sqlite3模块是Python数据持久化的基石。它轻量而强大，将SQLite数据库引擎无缝融入Python生态之中。掌握sqlite3不仅仅是学习一个模块的API，更是理解数据库交互的一般规律——连接管理、游标使用、参数化查询、事务控制，这些概念对后续学习其他数据库驱动（如psycopg2、mysql-connector-python）同样适用。扎实掌握sqlite3，等于为Python数据持久化能力打下了最坚实的基础。