第四十九篇：数据库操作——SQLite 与 MySQL
前言
几乎所有的应用都需要持久化存储数据，而数据库是最常见的方案。Python 标准库自带了 SQLite 支持，无需安装任何服务端即可使用；对于生产环境，MySQL 是最流行的关系型数据库之一。本文将系统讲解 SQLite 和 MySQL 的增删改查操作，以及 SQL 注入防护。
SQLite——零配置的轻量级数据库
为什么选 SQLite
┌───────────────────┐ ┌───────────────────┐
│ SQLite │ │ MySQL │
├───────────────────┤ ├───────────────────┤
│ 单文件存储 │ │ 服务端 + 客户端 │
│ 零配置，开箱即用 │ │ 需安装配置服务 │
│ 适合小型应用 │ │ 适合中大型应用 │
│ 支持事务 │ │ 支持事务 │
│ 并发读，单写 │ │ 高并发读写 │
└───────────────────┘ └───────────────────┘
SQLite 的核心优势：
- Python 内置
sqlite3模块，无需pip install - 整个数据库就是一个
.db文件，便于备份和迁移 - 支持 SQL 标准语法、事务、外键
- 适合原型开发、移动应用、桌面工具
连接数据库与建表
import sqlite3
# 连接数据库（文件不存在则自动创建）
conn = sqlite3.connect('test.db')
# 创建游标对象（执行 SQL 语句）
cursor = conn.cursor()
# 创建学生表
cursor.execute('''
CREATE TABLE IF NOT EXISTS students (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER,
grade TEXT DEFAULT 'A',
email TEXT UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建课程表（含外键）
cursor.execute('''
CREATE TABLE IF NOT EXISTS courses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
student_id INTEGER,
course_name TEXT NOT NULL,
score REAL,
FOREIGN KEY (student_id) REFERENCES students(id)
)
''')
conn.commit()
print("✓ 表创建成功")
增删改查（CRUD）
import sqlite3
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# ---------- 增（Create）----------
# 单条插入
cursor.execute(
"INSERT INTO students (name, age, grade, email) VALUES (?, ?, ?, ?)",
("张三", 20, "A", "zhangsan@example.com")
)
# 批量插入
students_data = [
("李四", 21, "B", "lisi@example.com"),
("王五", 19, "A", "wangwu@example.com"),
("赵六", 22, "C", "zhaoliu@example.com"),
("钱七", 20, "B", "qianqi@example.com"),
]
cursor.executemany(
"INSERT INTO students (name, age, grade, email) VALUES (?, ?, ?, ?)",
students_data
)
conn.commit()
print(f"✓ 插入了 {cursor.rowcount} 条记录")
# ---------- 查（Read）----------
# 查询所有记录
print("\n--- 所有学生 ---")
cursor.execute("SELECT * FROM students")
for row in cursor:
print(f"ID={row[0]}, 姓名={row[1]}, 年龄={row[2]}, 等级={row[3]}")
# 条件查询
print("\n--- 年龄 >= 20 的学生 ---")
cursor.execute("SELECT name, age, grade FROM students WHERE age >= ?", (20,))
for row in cursor:
print(f" {row[0]}: {row[1]}岁, 等级{row[2]}")
# 排序与限制
print("\n--- 按年龄降序前3名 ---")
cursor.execute("SELECT name, age FROM students ORDER BY age DESC LIMIT 3")
for row in cursor:
print(f" {row[0]}: {row[1]}岁")
# 聚合查询
print("\n--- 统计信息 ---")
cursor.execute("SELECT COUNT(*), AVG(age), MIN(age), MAX(age) FROM students")
count, avg_age, min_age, max_age = cursor.fetchone()
print(f" 总人数: {count}, 平均年龄: {avg_age:.1f}, 最小: {min_age}, 最大: {max_age}")
# 分组查询
print("\n--- 按等级分组 ---")
cursor.execute("SELECT grade, COUNT(*) as cnt FROM students GROUP BY grade ORDER BY cnt DESC")
for row in cursor:
print(f" 等级 {row[0]}: {row[1]} 人")
# ---------- 改（Update）----------
cursor.execute(
"UPDATE students SET grade = ? WHERE name = ?",
("A", "赵六")
)
conn.commit()
print(f"\n✓ 更新了 {cursor.rowcount} 条记录")
# ---------- 删（Delete）----------
cursor.execute("DELETE FROM students WHERE name = ?", ("钱七",))
conn.commit()
print(f"✓ 删除了 {cursor.rowcount} 条记录")
conn.close()
事务管理
import sqlite3
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 开启事务（sqlite3 默认自动开启事务）
try:
# 转账：张三给李四转 100 分
cursor.execute("UPDATE students SET age = age + 1 WHERE name = '张三'")
cursor.execute("UPDATE students SET age = age - 1 WHERE name = '李四'")
# 模拟异常
# raise Exception("模拟转账失败")
conn.commit()
print("✓ 事务提交成功")
except Exception as e:
conn.rollback()
print(f"❌ 事务回滚: {e}")
finally:
conn.close()
使用 Row 工厂（按列名访问）
import sqlite3
conn = sqlite3.connect('test.db')
# 设置行工厂，让查询结果可以通过列名访问
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM students WHERE grade = ?", ("A",))
print("--- 等级 A 的学生 ---")
for row in cursor:
# 通过列名访问，更直观
print(f" {row['name']} (ID: {row['id']}) - {row['age']}岁, 邮箱: {row['email']}")
conn.close()
封装数据库操作类
import sqlite3
from contextlib import contextmanager
class StudentDB:
"""学生数据库操作封装"""
def __init__(self, db_path='students.db'):
self.db_path = db_path
# 启用外键支持
with self._get_cursor() as cursor:
cursor.execute('''
CREATE TABLE IF NOT EXISTS students (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER,
grade TEXT DEFAULT 'A',
email TEXT UNIQUE
)
''')
@contextmanager
def _get_cursor(self):
"""上下文管理器：自动提交和关闭"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
yield cursor
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
conn.close()
def add_student(self, name, age, grade='A', email=None):
with self._get_cursor() as cursor:
cursor.execute(
"INSERT INTO students (name, age, grade, email) VALUES (?, ?, ?, ?)",
(name, age, grade, email)
)
return cursor.lastrowid
def get_student(self, student_id):
with self._get_cursor() as cursor:
cursor.execute("SELECT * FROM students WHERE id = ?", (student_id,))
row = cursor.fetchone()
return dict(row) if row else None
def get_all_students(self, grade=None):
with self._get_cursor() as cursor:
if grade:
cursor.execute("SELECT * FROM students WHERE grade = ?", (grade,))
else:
cursor.execute("SELECT * FROM students ORDER BY id")
return [dict(row) for row in cursor]
def update_student(self, student_id, **kwargs):
if not kwargs:
return False
fields = []
values = []
for key, value in kwargs.items():
fields.append(f"{key} = ?")
values.append(value)
values.append(student_id)
with self._get_cursor() as cursor:
cursor.execute(
f"UPDATE students SET {', '.join(fields)} WHERE id = ?",
values
)
return cursor.rowcount > 0
def delete_student(self, student_id):
with self._get_cursor() as cursor:
cursor.execute("DELETE FROM students WHERE id = ?", (student_id,))
return cursor.rowcount > 0
def search_students(self, keyword):
with self._get_cursor() as cursor:
cursor.execute(
"SELECT * FROM students WHERE name LIKE ? OR email LIKE ?",
(f"%{keyword}%", f"%{keyword}%")
)
return [dict(row) for row in cursor]
# 使用示例
if __name__ == '__main__':
db = StudentDB()
# 批量添加
ids = []
for name, age, grade, email in [
("孙悟空", 500, "S", "wukong@huaguo.com"),
("猪八戒", 300, "B", "bajie@tianzhu.com"),
("沙悟净", 400, "A", "wujing@liuhe.com"),
]:
sid = db.add_student(name, age, grade, email)
ids.append(sid)
print(f"✓ 添加: {name} -> ID={sid}")
# 查询所有
print("\n所有学生:")
for s in db.get_all_students():
print(f" {s['id']}. {s['name']} - {s['age']}岁, 等级{s['grade']}")
# 条件查询
print("\n等级 A 及以上:")
for s in db.get_all_students(grade='A'):
print(f" {s['name']}: {s['age']}岁")
# 搜索
print("\n搜索 '孙':")
for s in db.search_students("孙"):
print(f" {s['name']}")
# 更新
db.update_student(ids[1], grade='A', age=350)
print(f"\n✓ 更新猪八戒: 等级->A, 年龄->350")
# 验证
student = db.get_student(ids[1])
print(f" 当前: {student['name']}, {student['age']}岁, 等级{student['grade']}")
# 删除
db.delete_student(ids[2])
print(f"✓ 删除沙悟净")
MySQL——生产级关系型数据库
安装与连接
# 安装 MySQL 驱动
pip install PyMySQL
import pymysql
# 连接 MySQL
conn = pymysql.connect(
host='localhost',
port=3306,
user='root',
password='your_password',
database='test_db',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor # 返回字典格式
)
# 创建数据库（需要先连接 MySQL 服务器，不指定 database）
# conn = pymysql.connect(host='localhost', user='root', password='your_password')
# cursor.execute("CREATE DATABASE IF NOT EXISTS test_db CHARACTER SET utf8mb4")
cursor = conn.cursor()
print("✓ MySQL 连接成功")
MySQL 建表与操作
import pymysql
def get_connection():
return pymysql.connect(
host='localhost',
port=3306,
user='root',
password='your_password',
database='test_db',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
# 建表
with get_connection() as conn:
with conn.cursor() as cursor:
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
price DECIMAL(10, 2) NOT NULL,
stock INT DEFAULT 0,
category VARCHAR(50),
description TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
''')
conn.commit()
print("✓ products 表创建成功")
# 插入数据
with get_connection() as conn:
with conn.cursor() as cursor:
# 单条
cursor.execute(
"INSERT INTO products (name, price, stock, category) VALUES (%s, %s, %s, %s)",
("Python 编程书", 59.90, 100, "图书")
)
# 批量
products = [
("机械键盘", 299.00, 50, "电子产品"),
("无线鼠标", 89.90, 200, "电子产品"),
("保温杯", 39.90, 300, "日用品"),
("蓝牙耳机", 199.00, 80, "电子产品"),
("笔记本支架", 129.00, 150, "办公用品"),
]
cursor.executemany(
"INSERT INTO products (name, price, stock, category) VALUES (%s, %s, %s, %s)",
products
)
conn.commit()
print(f"✓ 插入 {len(products) + 1} 条产品")
# 查询数据
with get_connection() as conn:
with conn.cursor() as cursor:
# 模糊查询
cursor.execute(
"SELECT * FROM products WHERE name LIKE %s OR category = %s ORDER BY price DESC",
("%键%", "电子产品")
)
print("\n--- 电子产品或名称含'键' ---")
for row in cursor.fetchall():
print(f" {row['name']}: ¥{row['price']}, 库存{row['stock']}")
# 聚合查询
cursor.execute("""
SELECT category, COUNT(*) as count,
AVG(price) as avg_price,
SUM(stock) as total_stock
FROM products
GROUP BY category
ORDER BY avg_price DESC
""")
print("\n--- 按分类统计 ---")
for row in cursor.fetchall():
print(f" {row['category']}: {row['count']}个商品, "
f"均价¥{row['avg_price']:.2f}, 总库存{row['total_stock']}")
# 分页查询
page, size = 1, 3
offset = (page - 1) * size
cursor.execute(
"SELECT * FROM products ORDER BY id LIMIT %s OFFSET %s",
(size, offset)
)
print(f"\n--- 第{page}页（每页{size}条）---")
for row in cursor.fetchall():
print(f" [{row['id']}] {row['name']}")
SQL 注入与参数化查询
import pymysql
# ✗ 危险！直接拼接 SQL 字符串 —— 存在 SQL 注入漏洞
def unsafe_login(cursor, username, password):
# 攻击者输入 username = "' OR '1'='1" 即可绕过验证！
sql = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
cursor.execute(sql) # 危险！
return cursor.fetchone()
# ✓ 安全！使用参数化查询
def safe_login(cursor, username, password):
sql = "SELECT * FROM users WHERE username = %s AND password = %s"
cursor.execute(sql, (username, password)) # 安全
return cursor.fetchone()
# SQL 注入演示
print("""
=== SQL 注入原理 ===
恶意输入: username = ' OR '1'='1
拼接后 SQL: SELECT * FROM users WHERE username = '' OR '1'='1' AND password = 'xxx'
结果: WHERE 条件恒为真，返回所有用户！
修复方案: 使用参数化查询（占位符 ? 或 %s）
数据库驱动会自动转义特殊字符，杜绝注入。
""")
# 参数化查询的正确写法
def search_products(cursor, keyword, min_price=None, max_price=None):
"""安全的参数化查询示例"""
sql = "SELECT * FROM products WHERE name LIKE %s"
params = [f"%{keyword}%"]
if min_price is not None:
sql += " AND price >= %s"
params.append(min_price)
if max_price is not None:
sql += " AND price <= %s"
params.append(max_price)
sql += " ORDER BY price DESC"
cursor.execute(sql, params)
return cursor.fetchall()
# 使用
# results = search_products(cursor, "键盘", min_price=100)
MySQL 连接池
from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池（需安装：pip install dbutils）
pool = PooledDB(
creator=pymysql,
maxconnections=10, # 最大连接数
mincached=2, # 初始空闲连接数
maxcached=5, # 最大空闲连接数
host='localhost',
port=3306,
user='root',
password='your_password',
database='test_db',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
def query_products(category):
"""从连接池获取连接执行查询"""
conn = pool.connection()
try:
with conn.cursor() as cursor:
cursor.execute(
"SELECT * FROM products WHERE category = %s ORDER BY price",
(category,)
)
return cursor.fetchall()
finally:
conn.close() # 实际是归还到连接池
# 批量查询
for cat in ["电子产品", "图书", "日用品"]:
results = query_products(cat)
print(f"{cat}: {len(results)} 件商品")
SQLite 与 MySQL 对比实战
import sqlite3
import time
import os
def benchmark_sqlite():
"""SQLite 批量插入性能测试"""
db_path = 'benchmark.db'
if os.path.exists(db_path):
os.remove(db_path)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE benchmark_data (
id INTEGER PRIMARY KEY,
value REAL,
label TEXT
)
''')
# 逐条插入
start = time.time()
for i in range(1000):
cursor.execute(
"INSERT INTO benchmark_data (value, label) VALUES (?, ?)",
(i * 0.1, f"item_{i}")
)
conn.commit()
single_time = time.time() - start
# 清空重来
cursor.execute("DELETE FROM benchmark_data")
conn.commit()
# 批量插入
start = time.time()
cursor.executemany(
"INSERT INTO benchmark_data (value, label) VALUES (?, ?)",
[(i * 0.1, f"batch_{i}") for i in range(1000)]
)
conn.commit()
batch_time = time.time() - start
# 验证
cursor.execute("SELECT COUNT(*) FROM benchmark_data")
count = cursor.fetchone()[0]
conn.close()
os.remove(db_path)
print(f"SQLite 性能测试 ({count} 条记录):")
print(f" 逐条插入: {single_time:.3f}s")
print(f" 批量插入: {batch_time:.3f}s")
print(f" 批量快 {single_time / batch_time:.1f} 倍")
benchmark_sqlite()
选择建议
| 对比维度 | SQLite | MySQL |
|---|---|---|
| 安装 | 内置，零配置 | 需安装服务端 |
| 存储方式 | 单文件 | 服务端管理 |
| 并发能力 | 单写入 | 高并发读写 |
| 数据量 | GB 级别 | TB 级别 |
| 适用场景 | 原型、桌面应用、测试 | Web 应用、企业系统 |
| 用户管理 | 无 | 多用户权限 |
| 网络访问 | 本地文件 | TCP 网络 |
| 运维成本 | 几乎为零 | 需专业运维 |
实战项目：学生成绩管理系统
import sqlite3
class GradeManagementSystem:
"""学生成绩管理系统"""
def __init__(self, db_path='grades.db'):
self.db_path = db_path
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.executescript('''
CREATE TABLE IF NOT EXISTS students (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
class TEXT
);
CREATE TABLE IF NOT EXISTS subjects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
full_score REAL DEFAULT 100
);
CREATE TABLE IF NOT EXISTS scores (
id INTEGER PRIMARY KEY AUTOINCREMENT,
student_id INTEGER NOT NULL,
subject_id INTEGER NOT NULL,
score REAL NOT NULL,
exam_date TEXT DEFAULT (DATE('now')),
FOREIGN KEY (student_id) REFERENCES students(id),
FOREIGN KEY (subject_id) REFERENCES subjects(id),
UNIQUE(student_id, subject_id, exam_date)
);
''')
conn.commit()
conn.close()
def add_student(self, name, class_name):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
try:
cursor.execute(
"INSERT INTO students (name, class) VALUES (?, ?)",
(name, class_name)
)
conn.commit()
print(f"✓ 添加学生: {name} ({class_name})")
return cursor.lastrowid
except sqlite3.IntegrityError:
print(f"⚠ 学生 '{name}' 已存在")
return None
finally:
conn.close()
def add_subject(self, name, full_score=100):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
try:
cursor.execute(
"INSERT INTO subjects (name, full_score) VALUES (?, ?)",
(name, full_score)
)
conn.commit()
return cursor.lastrowid
except sqlite3.IntegrityError:
print(f"⚠ 科目 '{name}' 已存在")
return None
finally:
conn.close()
def record_score(self, student_name, subject_name, score):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 查找学生和科目 ID
cursor.execute("SELECT id FROM students WHERE name = ?", (student_name,))
student = cursor.fetchone()
if not student:
print(f"❌ 学生 '{student_name}' 不存在")
conn.close()
return
cursor.execute("SELECT id, full_score FROM subjects WHERE name = ?", (subject_name,))
subject = cursor.fetchone()
if not subject:
print(f"❌ 科目 '{subject_name}' 不存在")
conn.close()
return
if score > subject[1]:
print(f"❌ 分数 {score} 超过满分 {subject[1]}")
conn.close()
return
try:
cursor.execute(
"INSERT INTO scores (student_id, subject_id, score) VALUES (?, ?, ?)",
(student[0], subject[0], score)
)
conn.commit()
print(f"✓ 录入成绩: {student_name} {subject_name} = {score}分")
except sqlite3.IntegrityError:
print(f"⚠ {student_name} 的 {subject_name} 成绩已存在，使用更新")
cursor.execute(
"UPDATE scores SET score = ? WHERE student_id = ? AND subject_id = ?",
(score, student[0], subject[0])
)
conn.commit()
finally:
conn.close()
def get_student_report(self, student_name):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT s.name AS student, s.class,
sub.name AS subject, sc.score, sub.full_score,
ROUND(sc.score * 100.0 / sub.full_score, 1) AS percentage
FROM students s
JOIN scores sc ON sc.student_id = s.id
JOIN subjects sub ON sub.id = sc.subject_id
WHERE s.name = ?
ORDER BY sub.name
""", (student_name,))
rows = cursor.fetchall()
conn.close()
if not rows:
print(f"未找到学生 '{student_name}' 的成绩记录")
return
print(f"\n{'='*50}")
print(f" 学生成绩单: {rows[0]['student']} ({rows[0]['class']})")
print(f"{'='*50}")
total_score = 0
total_full = 0
for row in rows:
print(f" {row['subject']:>8s}: {row['score']:>6.1f} / {row['full_score']:.0f} "
f"({row['percentage']}%)")
total_score += row['score']
total_full += row['full_score']
avg = total_score / len(rows) if rows else 0
overall_pct = total_score * 100.0 / total_full if total_full else 0
print(f"{'-'*50}")
print(f" {'总分':>8s}: {total_score:>6.1f} / {total_full:.0f} ({overall_pct:.1f}%)")
print(f" {'平均分':>8s}: {avg:.1f}")
print(f"{'='*50}")
def class_ranking(self, subject_name=None):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
if subject_name:
cursor.execute("""
SELECT s.name AS student, s.class, sc.score
FROM students s
JOIN scores sc ON sc.student_id = s.id
JOIN subjects sub ON sub.id = sc.subject_id
WHERE sub.name = ?
ORDER BY sc.score DESC
""", (subject_name,))
rows = cursor.fetchall()
print(f"\n--- {subject_name} 成绩排名 ---")
for rank, row in enumerate(rows, 1):
print(f" {rank}. {row['student']} ({row['class']}): {row['score']}分")
else:
cursor.execute("""
SELECT s.name AS student, s.class,
SUM(sc.score) AS total_score,
AVG(sc.score) AS avg_score,
COUNT(sc.score) AS subject_count
FROM students s
LEFT JOIN scores sc ON sc.student_id = s.id
GROUP BY s.id
ORDER BY total_score DESC NULLS LAST
""")
rows = cursor.fetchall()
print(f"\n--- 总成绩排名 ---")
for rank, row in enumerate(rows, 1):
if row['total_score']:
print(f" {rank}. {row['student']} ({row['class']}): "
f"总分{row['total_score']:.1f}, 均分{row['avg_score']:.1f}")
else:
print(f" {rank}. {row['student']} ({row['class']}): 暂无成绩")
conn.close()
# ===== 运行系统 =====
if __name__ == '__main__':
system = GradeManagementSystem()
# 添加学生
system.add_student("刘备", "蜀汉班")
system.add_student("关羽", "蜀汉班")
system.add_student("张飞", "蜀汉班")
system.add_student("曹操", "魏国班")
system.add_student("孙权", "吴国班")
# 添加科目
system.add_subject("语文")
system.add_subject("数学")
system.add_subject("英语")
# 录入成绩
system.record_score("刘备", "语文", 95)
system.record_score("刘备", "数学", 88)
system.record_score("刘备", "英语", 92)
system.record_score("关羽", "语文", 98)
system.record_score("关羽", "数学", 90)
system.record_score("关羽", "英语", 85)
system.record_score("张飞", "语文", 80)
system.record_score("张飞", "数学", 75)
system.record_score("张飞", "英语", 82)
system.record_score("曹操", "语文", 90)
system.record_score("曹操", "数学", 95)
system.record_score("曹操", "英语", 88)
system.record_score("孙权", "语文", 85)
system.record_score("孙权", "数学", 82)
system.record_score("孙权", "英语", 90)
# 查看成绩单
system.get_student_report("关羽")
system.get_student_report("曹操")
# 排名
system.class_ranking("数学")
system.class_ranking()
小结
| 知识点 | 说明 |
|---|---|
sqlite3 模块 |
Python 内置，无需安装，单文件存储 |
pymysql 模块 |
连接 MySQL 的纯 Python 驱动 |
| 参数化查询 | 使用 ? 或 %s 占位符，防止 SQL 注入 |
| 事务管理 | commit() 提交，rollback() 回滚 |
executemany() |
批量操作，性能远超逐条执行 |
| 连接池 | 复用连接，减少开销，适合高并发场景 |
| 外键约束 | 保证数据完整性，关联表一致性 |
选择建议：开发和测试阶段用 SQLite，上线部署用 MySQL。Python 的 DB-API 2.0 接口标准让两者代码结构几乎一致，迁移成本低。
练习题
使用 SQLite 创建一个图书管理数据库，包含图书表（书名、作者、价格、库存）和借阅记录表（读者、图书、借阅日期、归还日期），实现借书和还书功能。
编写一个函数，使用参数化查询从 MySQL 中按多个条件搜索商品（分类、价格范围、关键词），返回排序后的结果列表。
为成绩管理系统增加"删除学生"功能，注意处理外键约束——当学生有成绩记录时，应先删除关联成绩再删除学生，或使用
ON DELETE CASCADE外键策略。