Python SQLite 数据库使用指南
Python SQLite 详解
SQLite 是一个轻量级的嵌入式数据库，Python 通过 sqlite3 模块提供了对 SQLite 数据库的支持。以下是详细的使用指南。
目录
基本概念
SQLite 的特点：
- 无需单独的服务器进程
- 数据库存储在单个磁盘文件中
- 零配置
- 支持大多数 SQL 标准
连接数据库
import sqlite3
# 连接到数据库（如果不存在则创建）
conn = sqlite3.connect('example.db')
# 创建游标对象
cursor = conn.cursor()
# 关闭连接
conn.close()
使用上下文管理器自动关闭连接：
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 执行操作
创建表
import sqlite3
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 创建用户表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建订单表（外键示例）
cursor.execute('''
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
product TEXT NOT NULL,
price REAL NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
conn.commit()
插入数据
单条插入
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 插入单条数据
cursor.execute('''
INSERT INTO users (name, email, age)
VALUES (?, ?, ?)
''', ('张三', 'zhangsan@email.com', 25))
# 获取最后插入的ID
last_id = cursor.lastrowid
print(f"最后插入的ID: {last_id}")
conn.commit()
批量插入
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
users = [
('李四', 'lisi@email.com', 30),
('王五', 'wangwu@email.com', 28),
('赵六', 'zhaoliu@email.com', 35)
]
cursor.executemany('''
INSERT INTO users (name, email, age)
VALUES (?, ?, ?)
''', users)
conn.commit()
查询数据
基本查询
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 查询所有数据
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
print(row)
# 查询单条数据
cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
user = cursor.fetchone()
print(f"用户1: {user}")
# 查询特定字段
cursor.execute('SELECT name, email FROM users WHERE age > ?', (25,))
users = cursor.fetchall()
for user in users:
print(f"姓名: {user[0]}, 邮箱: {user[1]}")
使用字典格式返回结果
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
with sqlite3.connect('example.db') as conn:
conn.row_factory = dict_factory
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
for user in users:
print(f"ID: {user['id']}, 姓名: {user['name']}, 邮箱: {user['email']}")
更新和删除数据
更新数据
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 更新数据
cursor.execute('''
UPDATE users
SET age = ?, email = ?
WHERE id = ?
''', (26, 'new_email@example.com', 1))
# 检查受影响的行数
print(f"更新了 {cursor.rowcount} 行")
conn.commit()
删除数据
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 删除数据
cursor.execute('DELETE FROM users WHERE id = ?', (3,))
print(f"删除了 {cursor.rowcount} 行")
conn.commit()
事务处理
with sqlite3.connect('example.db') as conn:
try:
cursor = conn.cursor()
# 开始事务（SQLite 默认自动提交关闭）
cursor.execute('BEGIN TRANSACTION')
# 执行多个操作
cursor.execute('INSERT INTO users (name, email, age) VALUES (?, ?, ?)',
('测试用户', 'test@email.com', 40))
cursor.execute('UPDATE users SET age = ? WHERE name = ?', (41, '测试用户'))
# 提交事务
conn.commit()
print("事务提交成功")
except sqlite3.Error as e:
# 回滚事务
conn.rollback()
print(f"事务回滚: {e}")
使用上下文管理器
class Database:
def __init__(self, db_name):
self.db_name = db_name
def __enter__(self):
self.conn = sqlite3.connect(self.db_name)
self.conn.row_factory = sqlite3.Row
return self.conn.cursor()
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.conn.commit()
else:
self.conn.rollback()
self.conn.close()
# 使用自定义上下文管理器
with Database('example.db') as cursor:
cursor.execute('SELECT * FROM users')
for row in cursor:
print(dict(row))
错误处理
import sqlite3
try:
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 尝试插入重复的email（违反UNIQUE约束）
cursor.execute('''
INSERT INTO users (name, email, age)
VALUES (?, ?, ?)
''', ('重复用户', 'zhangsan@email.com', 25))
conn.commit()
except sqlite3.IntegrityError as e:
print(f"完整性错误: {e}")
except sqlite3.Error as e:
print(f"数据库错误: {e}")
finally:
print("操作完成")
高级功能
使用索引提高查询性能
with sqlite3.connect('example.db') as conn:
cursor = conn.cursor()
# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_email ON users (email)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_age ON users (age)')
conn.commit()
备份数据库
def backup_database(source_db, backup_db):
with sqlite3.connect(source_db) as source:
with sqlite3.connect(backup_db) as backup:
source.backup(backup)
backup_database('example.db', 'example_backup.db')
使用自定义聚合函数
class Average:
def __init__(self):
self.count = 0
self.total = 0
def step(self, value):
if value is not None:
self.count += 1
self.total += value
def finalize(self):
return self.total / self.count if self.count > 0 else 0
with sqlite3.connect('example.db') as conn:
conn.create_aggregate("my_avg", 1, Average)
cursor = conn.cursor()
cursor.execute('SELECT my_avg(age) FROM users')
result = cursor.fetchone()
print(f"平均年龄: {result[0]}")
最佳实践
- 使用参数化查询防止SQL注入
- 始终关闭连接或使用上下文管理器
- 合理使用事务提高性能和数据一致性
- 创建适当的索引优化查询性能
- **使用
IF NOT EXISTS**避免重复创建表 - 处理异常确保数据库完整性
- 定期备份重要数据
# 完整示例
import sqlite3
from contextlib import contextmanager
@contextmanager
def get_db_connection(db_name):
"""数据库连接上下文管理器"""
conn = sqlite3.connect(db_name)
conn.row_factory = sqlite3.Row
try:
yield conn
except sqlite3.Error:
conn.rollback()
raise
finally:
conn.close()
def init_database():
"""初始化数据库"""
with get_db_connection('example.db') as conn:
cursor = conn.cursor()
# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_email ON users (email)')
conn.commit()
if __name__ == "__main__":
init_database()
# 使用示例
with get_db_connection('example.db') as conn:
cursor = conn.cursor()
# 插入数据
cursor.execute(
'INSERT OR IGNORE INTO users (name, email, age) VALUES (?, ?, ?)',
('测试用户', 'test@example.com', 30)
)
# 查询数据
cursor.execute('SELECT * FROM users')
for row in cursor:
print(dict(row))
conn.commit()
这个详细的指南涵盖了 Python 中使用 SQLite 的主要方面，从基础操作到高级功能，希望能帮助你更好地使用 SQLite 数据库。
