上一篇
Python结合SQLite，先导入sqlite3库，创建连接与游标，执行建表语句如
CREATE TABLE tb(id INT PRIMARY KEY, name TEXT)，再增删改查
理解需求与选择工具
要创建一个“简单”的数据库系统，首先需明确目标：存储结构化数据（如用户信息、商品清单）、支持增删改查操作（CRUD），对于初学者而言，推荐使用轻量级关系型数据库管理系统（RDBMS），SQLite（无需安装服务器，文件即数据库）或 MySQL/PostgreSQL（适合进阶练习），这里以 Python + SQLite 组合为例,因其跨平台且语法直观。
关键技术栈对比表：
| 组件 | 推荐方案 | 优势 |
|---|---|---|
| 编程语言 | Python | 语法简洁，生态丰富（如Pandas） |
| 数据库引擎 | SQLite | 零配置，嵌入到应用程序中 |
| ORM框架 | 原生SQL / peewee库 | 直接控制SQL或简化对象映射 |
环境准备与基础架构搭建
安装依赖库
若使用Python开发，需确保已安装sqlite3模块（Python标准库自带）,可选装第三方库提升效率：
pip install pandas # 用于数据分析辅助验证结果 pip install prettytable # 美化终端输出表格展示
设计数据模型
假设我们要管理一本书籍收藏夹，核心字段包括：ID（主键）、书名、作者、出版年份、评分,对应的SQL建表语句如下：
CREATE TABLE IF NOT EXISTS books (
id INTEGER PRIMARY KEY AUTOINCREMENT,TEXT NOT NULL,
author TEXT,
year_published INTEGER,
rating REAL CHECK(rating >= 0 AND rating <= 10)
);
- 约束说明：
NOT NULL确保必填项；CHECK限制评分范围为[0,10]。
核心功能实现代码详解
以下是完整的Python脚本示例，包含连接数据库、执行CRUD操作及异常处理：
import sqlite3
from contextlib import closing
import sys
class BookDatabase:
def __init__(self, db_name='library.db'):
self.conn = None
self.db_path = db_name
self._initialize_schema()
def _initialize_schema(self):
"""首次运行时自动创建表结构"""
with closing(sqlite3.connect(self.db_path)) as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS books (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
author TEXT,
year_published INTEGER,
rating REAL CHECK(rating >= 0 AND rating <= 10)
);
''')
conn.commit()
def add_book(self, title, author, year, rating):
"""插入新书记录"""
try:
with closing(sqlite3.connect(self.db_path)) as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO books (title, author, year_published, rating) VALUES (?, ?, ?, ?)",
(title, author, year, float(rating))
)
conn.commit()
return cursor.lastrowid # 返回自增ID
except ValueError as e:
print(f"输入错误: {str(e)}", file=sys.stderr)
return None
except sqlite3.IntegrityError as e:
print(f"违反唯一性约束或其他完整性规则: {str(e)}", file=sys.stderr)
return None
def delete_book(self, book_id):
"""根据ID删除指定书籍"""
affected_rows = 0
with closing(sqlite3.connect(self.db_path)) as conn:
cursor = conn.cursor()
affected_rows = cursor.execute(
"DELETE FROM books WHERE id = ?", (book_id,)
).rowcount
conn.commit()
return affected_rows > 0
def update_book(self, book_id, new_title=None, new_author=None, new_year=None, new_rating=None):
"""动态更新部分字段"""
params = []
set_clauses = []
if new_title is not None:
set_clauses.append("title = ?")
params.append(new_title)
if new_author is not None:
set_clauses.append("author = ?")
params.append(new_author)
if new_year is not None:
set_clauses.append("year_published = ?")
params.append(new_year)
if new_rating is not None:
set_clauses.append("rating = ?")
params.append(float(new_rating))
if not set_clauses:
raise ValueError("至少提供一个可更新的字段")
sql = f"UPDATE books SET {', '.join(set_clauses)} WHERE id = ?"
params.append(book_id)
with closing(sqlite3.connect(self.db_path)) as conn:
cursor = conn.cursor()
cursor.execute(sql, params)
conn.commit()
return cursor.rowcount == 1
def list_all_books(self):
"""查询所有书籍并以易读格式返回"""
result = []
with closing(sqlite3.connect(self.db_path)) as conn:
conn.row_factory = sqlite3.Row # 允许按列名访问行数据
cursor = conn.cursor()
for row in cursor.execute("SELECT FROM books ORDER BY id"):
result.append({
'ID': row['id'],
'书名': row['title'],
'作者': row['author'],
'出版年份': row['year_published'],
'评分': round(row['rating'], 1)
})
return result
# 测试用例演示
if __name__ == "__main__":
db = BookDatabase()
# 添加示例数据
db.add_book("三体", "刘慈欣", 2006, 9.5)
db.add_book("百年孤独", "加西亚·马尔克斯", 1967, 9.8)
# 更新第一条记录的评分
db.update_book(1, new_rating=9.7)
# 删除第二条记录
db.delete_book(2)
# 打印当前所有书籍
books = db.list_all_books()
from prettytable importPrettyTable
tbl = PrettyTable(["ID", "书名", "作者", "出版年份", "评分"])
for b in books:
tbl.add_row([b['ID'], b['书名'], b['作者'], b['出版年份'], b['评分']])
print(tbl)
代码逐行解析：
- 类初始化：
BookDatabase类封装了所有数据库交互逻辑，构造函数自动调用_initialize_schema()方法确保表存在。 - 事务管理：使用
with closing()上下文管理器自动提交/回滚事务,避免手动处理连接关闭问题。 - 参数化查询：所有SQL语句均使用问号占位符（），防止SQL注入攻击，例如插入操作中的
VALUES (?, ?, ?, ?)。 - 动态更新逻辑：
update_book()方法允许只修改部分字段,通过条件判断构建SET子句。 - 错误处理：捕获类型转换异常（如非数字的评分）、主键冲突等常见错误并给出友好提示。
- 结果格式化：利用
sqlite3.Row工厂将查询结果转为字典形式，配合prettytable库生成美观的控制台表格。
扩展建议与最佳实践
- 索引优化：频繁查询的列应建立索引以提高性能，例如为
title添加全文搜索索引：CREATE INDEX idx_title ON books(title);
- 备份机制：定期导出SQL脚本或复制物理文件进行备份，SQLite可通过命令行工具实现：
sqlite3 library.db .dump > backup_$(date +%Y%m%d).sql
- 迁移工具：生产环境中推荐使用Alembic等版本控制工具管理数据库结构变更。
- 安全加固：限制用户权限（如只读账户）、启用WAL模式提升并发写入能力。
FAQs常见问题解答
Q1: 如果运行代码时出现“OperationalError: unable to open database file”，该怎么办？
A: 此错误通常由以下原因导致：①文件路径无写入权限（检查目录是否存在且用户有读写权）；②跨设备链接符号链接失效（确保数据库文件与应用程序在同一磁盘分区），解决方案包括：切换到绝对路径指定数据库位置，或手动创建空文件后赋予适当权限，例如在Linux下执行：touch /path/to/library.db && chmod 664 /path/to/library.db。
Q2: 如何将现有CSV文件中的数据批量导入到这个数据库中？
A: 可以使用Python的csv模块结合事务批量插入,示例代码如下：
import csv
with open('books.csv', 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
with closing(sqlite3.connect(self.db_path)) as conn:
cursor = conn.cursor()
for row in reader:
cursor.execute('''
INSERT INTO books (title, author, year_published, rating)
VALUES (?, ?, ?, ?)
''', (row['title'], row['author'], int(row['year']), float(row['rating'])))
conn.commit()
注意需确保CSV列名与数据库字段完全匹配，并对数值类型做显式转换
