🐍 Python Day46：Python 操作 SQLite — 用代码管理数据库

🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握 sqlite3 模块、连接/游标、参数化查询、事务

📖 今日目录

1. SQLite 简介
2. 连接数据库与游标
3. 创建表
4. 插入数据 (INSERT)
5. 查询数据 (SELECT)
6. 参数化查询 — 防 SQL 注入
7. 更新与删除
8. 事务 (Transaction)
9. 上下文管理器
10. 实战：通讯录管理
11. 今日小结

1. SQLite 简介

SQLite 是 Python 内置的轻量级数据库——整个数据库就是一个文件，无需安装服务器。

# SQLite 的优势
# 1. Python 内置，不需要额外安装
# 2. 整个数据库就是一个 .db 文件
# 3. 适合小型应用、本地存储、原型开发
# 4. 支持标准 SQL 语法

# 适用场景
# ✅ 本地数据存储（配置、缓存、日志）
# ✅ 小型应用（<10万条数据）
# ✅ 原型开发和测试
# ❌ 高并发写入（只支持一个写入者）
# ❌ 大规模数据（>1GB 建议用 MySQL/PostgreSQL）

2. 连接数据库与游标

import sqlite3

# 连接数据库（文件不存在会自动创建）
conn = sqlite3.connect("mydata.db")

# 创建游标（执行 SQL 的工具）
cursor = conn.cursor()

# 执行 SQL
cursor.execute("SELECT sqlite_version()")
version = cursor.fetchone()
print(f"SQLite 版本: {version[0]}")

# 关闭连接
cursor.close()
conn.close()

# 获取查询结果的三种方式
import sqlite3

conn = sqlite3.connect("mydata.db")
cursor = conn.cursor()

cursor.execute("SELECT name, age FROM students")

# 方式1: fetchone() — 取一行
row = cursor.fetchone()
print(row)  # ('张三', 20)

# 方式2: fetchall() — 取所有行
rows = cursor.fetchall()
print(rows)  # [('张三', 20), ('李四', 22), ...]

# 方式3: 直接迭代游标
for row in cursor.execute("SELECT name, age FROM students"):
    print(f"{row[0]}: {row[1]}岁")

conn.close()

⚠️ 记住关闭连接！不关闭会导致数据库文件锁定，其他程序无法访问。推荐用 with 语句（后面会讲）。

3. 创建表

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# 创建学生表
cursor.execute("""
CREATE TABLE IF NOT EXISTS students (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    age INTEGER CHECK(age > 0 AND age < 150),
    gender TEXT DEFAULT '未知',
    class TEXT NOT NULL,
    email TEXT UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")

# 创建成绩表
cursor.execute("""
CREATE TABLE IF NOT EXISTS scores (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    student_id INTEGER NOT NULL,
    subject TEXT NOT NULL,
    score REAL CHECK(score >= 0 AND score <= 100),
    FOREIGN KEY (student_id) REFERENCES students(id)
)
""")

conn.commit()  # 提交更改
conn.close()

print("✅ 表创建成功")

4. 插入数据 (INSERT)

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# 插入单条数据
cursor.execute(
    "INSERT INTO students (name, age, gender, class) VALUES (?, ?, ?, ?)",
    ("张三", 20, "男", "一班")
)

# 插入多条数据
students = [
    ("李四", 22, "女", "二班"),
    ("王五", 21, "男", "一班"),
    ("赵六", 23, "男", "二班"),
    ("钱七", 20, "女", "一班"),
]
cursor.executemany(
    "INSERT INTO students (name, age, gender, class) VALUES (?, ?, ?, ?)",
    students
)

# 获取最后插入的 ID
print(f"最后插入的 ID: {cursor.lastrowid}")

conn.commit()
conn.close()
print(f"✅ 插入 {len(students) + 1} 条数据")

5. 查询数据 (SELECT)

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# 查询所有
cursor.execute("SELECT * FROM students")
all_students = cursor.fetchall()
for s in all_students:
    print(f"  ID={s[0]} | {s[1]} | {s[2]}岁 | {s[3]} | {s[4]}班")

# 查询单条
cursor.execute("SELECT name, age FROM students WHERE id = ?", (1,))
student = cursor.fetchone()
print(f"\n查询 ID=1: {student}")

# 查询并排序
cursor.execute("SELECT name, age FROM students ORDER BY age DESC")
for row in cursor.fetchall():
    print(f"  {row[0]}: {row[1]}岁")

# 查询并统计
cursor.execute("SELECT COUNT(*) FROM students")
count = cursor.fetchone()[0]
print(f"\n总人数: {count}")

# 分页查询
page = 1
page_size = 2
offset = (page - 1) * page_size
cursor.execute("SELECT name FROM students LIMIT ? OFFSET ?", (page_size, offset))
print(f"\n第{page}页: {cursor.fetchall()}")

conn.close()

6. 参数化查询 — 防 SQL 注入

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# ❌ 危险！字符串拼接（SQL 注入风险）
user_input = "张三' OR '1'='1"
query = f"SELECT * FROM students WHERE name = '{user_input}'"
print(f"恶意查询: {query}")
# SELECT * FROM students WHERE name = '张三' OR '1'='1'
# 这会返回所有数据！
cursor.execute(query)  # 千万不要这样做！

# ✅ 安全！参数化查询
cursor.execute("SELECT * FROM students WHERE name = ?", (user_input,))
result = cursor.fetchall()
print(f"安全查询结果: {result}")  # []（没有匹配，因为没有叫那个名字的人）

# 参数化查询用 ? 占位符，数据库会自动转义参数
# 绝对安全，不会被 SQL 注入

conn.close()

⚠️ 永远不要用字符串拼接构造 SQL！
❌ f"SELECT * FROM users WHERE name = '{name}'"
✅ cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

7. 更新与删除

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# 更新数据
cursor.execute(
    "UPDATE students SET age = ? WHERE name = ?",
    (21, "张三")
)
print(f"更新了 {cursor.rowcount} 行")  # rowcount = 受影响的行数

# 删除数据
cursor.execute("DELETE FROM students WHERE name = ?", ("赵六",))
print(f"删除了 {cursor.rowcount} 行")

conn.commit()

# 验证
cursor.execute("SELECT name, age FROM students")
for row in cursor.fetchall():
    print(f"  {row[0]}: {row[1]}岁")

conn.close()

8. 事务 (Transaction)

import sqlite3

conn = sqlite3.connect("school.db")
cursor = conn.cursor()

# 事务：要么全部成功，要么全部回滚
try:
    # 转账：张三 -100，李四 +100
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE name = '张三'")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE name = '李四'")

    # 两条都成功才提交
    conn.commit()
    print("✅ 转账成功")
except Exception as e:
    # 任何一条失败就回滚
    conn.rollback()
    print(f"❌ 转账失败，已回滚: {e}")

# SQLite 默认自动提交（autocommit）
# 用 conn.commit() 手动提交事务
# 用 conn.rollback() 回滚事务

💡 事务的 ACID 特性：
• Atomicity（原子性）：要么全部成功，要么全部失败
• Consistency（一致性）：数据始终保持合法状态
• Isolation（隔离性）：并发事务互不干扰
• Durability（持久性）：提交后数据永久保存

9. 上下文管理器

import sqlite3

# ✅ 推荐：用 with 语句自动管理连接
with sqlite3.connect("school.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM students")
    print(cursor.fetchall())
# with 块结束时自动 commit，出错时自动 rollback

# 封装一个数据库操作类
class Database:
    def __init__(self, db_path):
        self.db_path = db_path

    def __enter__(self):
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row  # 让查询结果像字典一样访问
        return self.conn.cursor()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.conn.rollback()
        else:
            self.conn.commit()
        self.conn.close()

# 使用
with Database("school.db") as db:
    db.execute("SELECT name, age FROM students")
    for row in db.fetchall():
        print(f"  {row['name']}: {row['age']}岁")  # 可以用列名访问

10. 实战：通讯录管理

import sqlite3

class ContactDB:
    """通讯录数据库管理"""

    def __init__(self, db_path="contacts.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self.cursor = self.conn.cursor()
        self._create_table()

    def _create_table(self):
        self.cursor.execute("""
        CREATE TABLE IF NOT EXISTS contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            phone TEXT,
            email TEXT,
            group_name TEXT DEFAULT '默认',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        self.conn.commit()

    def add(self, name, phone="", email="", group="默认"):
        """添加联系人"""
        self.cursor.execute(
            "INSERT INTO contacts (name, phone, email, group_name) VALUES (?, ?, ?, ?)",
            (name, phone, email, group)
        )
        self.conn.commit()
        return self.cursor.lastrowid

    def search(self, keyword):
        """搜索联系人"""
        self.cursor.execute(
            "SELECT * FROM contacts WHERE name LIKE ? OR phone LIKE ?",
            (f"%{keyword}%", f"%{keyword}%")
        )
        return self.cursor.fetchall()

    def get_all(self, group=None):
        """获取所有联系人"""
        if group:
            self.cursor.execute("SELECT * FROM contacts WHERE group_name = ?", (group,))
        else:
            self.cursor.execute("SELECT * FROM contacts ORDER BY name")
        return self.cursor.fetchall()

    def update(self, contact_id, **kwargs):
        """更新联系人"""
        allowed = {"name", "phone", "email", "group_name"}
        updates = {k: v for k, v in kwargs.items() if k in allowed and v is not None}
        if not updates:
            return

        set_clause = ", ".join(f"{k} = ?" for k in updates)
        values = list(updates.values()) + [contact_id]
        self.cursor.execute(f"UPDATE contacts SET {set_clause} WHERE id = ?", values)
        self.conn.commit()
        return self.cursor.rowcount

    def delete(self, contact_id):
        """删除联系人"""
        self.cursor.execute("DELETE FROM contacts WHERE id = ?", (contact_id,))
        self.conn.commit()
        return self.cursor.rowcount

    def close(self):
        self.conn.close()

# 使用
db = ContactDB()

# 添加联系人
db.add("张三", "13800138001", "zhangsan@example.com", "朋友")
db.add("李四", "13800138002", "lisi@example.com", "同事")
db.add("王五", "13800138003", group="朋友")
db.add("赵六", "13800138004", "zhaoliu@example.com")

# 搜索
print("🔍 搜索 '张三':")
for c in db.search("张三"):
    print(f"  {c['name']} | {c['phone']} | {c['email']}")

# 按组查看
print("\n👥 '朋友' 组:")
for c in db.get_all("朋友"):
    print(f"  {c['name']} | {c['phone']}")

# 更新
db.update(1, phone="13900139000")
print(f"\n✏️ 更新张三电话: 修改了 {db.update(1, phone='13900139000')} 行")

# 删除
db.delete(3)
print(f"\n🗑️ 删除王五: 修改了 {db.delete(3)} 行")

# 查看所有
print("\n📋 所有联系人:")
for c in db.get_all():
    print(f"  [{c['group_name']}] {c['name']} | {c['phone']} | {c['email']}")

db.close()

11. 今日小结

🎯 练习建议：
1. 给通讯录添加"分页查询"和"按组统计"功能
2. 用 SQLite 实现一个"每日记事本"（按日期存储和查询）
3. 把之前爬取的豆瓣 Top250 CSV 导入 SQLite 数据库

📚 Day46 完成！明天学习 Python 操作 MySQL