数据库编程基础：SQLite3、CRUD与参数化查询

数据库是现代应用的核心组件，用于持久化存储和管理数据。Python内置了sqlite3模块，让我们无需安装额外软件就能学习数据库编程。本文将从基础开始，带你掌握数据库操作的核心技能。

SQLite简介

SQLite是一个轻量级的嵌入式数据库，不需要服务器进程，数据存储在单个文件中。它是Python内置的，非常适合学习和小型应用。

import sqlite3

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

# 获取游标对象
cursor = conn.cursor()

# 执行SQL语句
cursor.execute("SELECT sqlite_version()")

# 获取结果
version = cursor.fetchone()
print(f"SQLite版本: {version[0]}")

# 关闭连接
conn.close()

创建表

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,
        grade TEXT,
        score REAL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# 创建课程表
cursor.execute("""
    CREATE TABLE IF NOT EXISTS courses (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        teacher TEXT,
        credits INTEGER DEFAULT 3
    )
""")

# 创建选课表（多对多关系）
cursor.execute("""
    CREATE TABLE IF NOT EXISTS enrollments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        student_id INTEGER,
        course_id INTEGER,
        grade TEXT,
        FOREIGN KEY (student_id) REFERENCES students(id),
        FOREIGN KEY (course_id) REFERENCES courses(id)
    )
""")

# 提交事务
conn.commit()
conn.close()

CRUD操作

创建（Create）

import sqlite3

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

# 插入单条记录
cursor.execute("""
    INSERT INTO students (name, age, grade, score)
    VALUES (?, ?, ?, ?)
""", ("张三", 20, "A", 85.5))

# 获取刚插入的ID
student_id = cursor.lastrowid
print(f"插入的学生ID: {student_id}")

# 插入多条记录
students = [
    ("李四", 21, "B", 78.0),
    ("王五", 19, "A", 92.5),
    ("赵六", 22, "C", 65.0),
    ("钱七", 20, "B", 88.0)
]

cursor.executemany("""
    INSERT INTO students (name, age, grade, score)
    VALUES (?, ?, ?, ?)
""", students)

# 插入课程数据
courses = [
    ("Python编程", "张老师", 4),
    ("数据库原理", "李老师", 3),
    ("计算机网络", "王老师", 3)
]
cursor.executemany("""
    INSERT INTO courses (name, teacher, credits)
    VALUES (?, ?, ?)
""", courses)

conn.commit()
print(f"插入了 {cursor.rowcount} 条记录")
conn.close()

读取（Read）

import sqlite3

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

# 查询所有记录
cursor.execute("SELECT * FROM students")
all_students = cursor.fetchall()
for student in all_students:
    print(student)

# 查询单条记录
cursor.execute("SELECT * FROM students WHERE name = ?", ("张三",))
student = cursor.fetchone()
print(f"找到学生: {student}")

# 查询多条记录
cursor.execute("SELECT * FROM students WHERE score > ?", (80,))
good_students = cursor.fetchall()
print(f"成绩大于80的学生: {len(good_students)} 人")

# 条件查询
cursor.execute("""
    SELECT name, age, score 
    FROM students 
    WHERE age >= 20 AND grade IN ('A', 'B')
    ORDER BY score DESC
""")
results = cursor.fetchall()
for name, age, score in results:
    print(f"{name} ({age}岁) - 成绩: {score}")

# 聚合查询
cursor.execute("""
    SELECT 
        COUNT(*) as total,
        AVG(score) as avg_score,
        MAX(score) as max_score,
        MIN(score) as min_score
    FROM students
""")
stats = cursor.fetchone()
print(f"总人数: {stats[0]}, 平均分: {stats[1]:.1f}")

更新（Update）

import sqlite3

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

# 更新单条记录
cursor.execute("""
    UPDATE students 
    SET score = ?, grade = ?
    WHERE name = ?
""", (95.0, "A+", "张三"))
print(f"更新了 {cursor.rowcount} 条记录")

# 批量更新
cursor.execute("""
    UPDATE students 
    SET score = score + 5
    WHERE score < 60
""")
print(f"加分了 {cursor.rowcount} 名学生")

# 条件更新
cursor.execute("""
    UPDATE students 
    SET grade = CASE 
        WHEN score >= 90 THEN 'A'
        WHEN score >= 80 THEN 'B'
        WHEN score >= 70 THEN 'C'
        WHEN score >= 60 THEN 'D'
        ELSE 'F'
    END
""")

conn.commit()
conn.close()

删除（Delete）

import sqlite3

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

# 删除单条记录
cursor.execute("DELETE FROM students WHERE name = ?", ("赵六",))
print(f"删除了 {cursor.rowcount} 条记录")

# 批量删除
cursor.execute("DELETE FROM students WHERE score < 60")

# 清空表
cursor.execute("DELETE FROM students")  # 保留表结构
# 或者
# cursor.execute("DROP TABLE students")  # 删除表

conn.commit()
conn.close()

参数化查询与防SQL注入

import sqlite3

def safe_query(conn, query, params=None):
    """安全的查询封装"""
    cursor = conn.cursor()
    try:
        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)
        return cursor.fetchall()
    except sqlite3.Error as e:
        print(f"查询错误: {e}")
        return []

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

# 正确：使用参数化查询
name = "张三"
cursor = conn.cursor()
cursor.execute("SELECT * FROM students WHERE name = ?", (name,))
result = cursor.fetchone()

# 错误：字符串拼接（易受SQL注入攻击）
# cursor.execute(f"SELECT * FROM students WHERE name = '{name}'")  # 不要这样做！

# 动态构建查询
def dynamic_search(conn, **kwargs):
    """动态条件搜索"""
    conditions = []
    params = []
    
    for key, value in kwargs.items():
        if value is not None:
            conditions.append(f"{key} = ?")
            params.append(value)
    
    if not conditions:
        return []
    
    where_clause = " AND ".join(conditions)
    query = f"SELECT * FROM students WHERE {where_clause}"
    
    return safe_query(conn, query, params)

# 使用动态搜索
results = dynamic_search(conn, grade="A", age=20)
print(f"搜索结果: {len(results)} 条")

conn.close()

事务管理

import sqlite3

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

try:
    # 开始事务（sqlite3默认自动开启）
    cursor = conn.cursor()
    
    # 执行多个操作
    cursor.execute("""
        INSERT INTO students (name, age, grade, score)
        VALUES (?, ?, ?, ?)
    """, ("新学生", 20, "A", 85))
    
    student_id = cursor.lastrowid
    
    cursor.execute("""
        INSERT INTO enrollments (student_id, course_id)
        VALUES (?, ?)
    """, (student_id, 1))
    
    # 提交事务
    conn.commit()
    print("事务提交成功")
    
except sqlite3.Error as e:
    # 发生错误，回滚事务
    conn.rollback()
    print(f"事务回滚: {e}")
    
finally:
    conn.close()

# 使用上下文管理器
with sqlite3.connect("school.db") as conn:
    conn.execute("PRAGMA foreign_keys = ON")
    cursor = conn.cursor()
    
    # 使用with语句自动提交/回滚
    with conn:
        cursor.execute("""
            INSERT INTO students (name, age, grade, score)
            VALUES (?, ?, ?, ?)
        """, ("自动提交学生", 19, "B", 78))

连接池与性能优化

import sqlite3
from contextlib import contextmanager

class Database:
    def __init__(self, db_path):
        self.db_path = db_path
    
    @contextmanager
    def connection(self):
        """获取数据库连接的上下文管理器"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row  # 使用字典式访问
        try:
            yield conn
        finally:
            conn.close()
    
    def execute(self, query, params=None):
        with self.connection() as conn:
            cursor = conn.cursor()
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            return cursor.fetchall()
    
    def executemany(self, query, params_list):
        with self.connection() as conn:
            cursor = conn.cursor()
            cursor.executemany(query, params_list)
            conn.commit()
    
    def fetch_one(self, query, params=None):
        results = self.execute(query, params)
        return results[0] if results else None

# 使用示例
db = Database("school.db")

# 使用Row工厂可以按列名访问
results = db.execute("SELECT * FROM students WHERE grade = ?", ("A",))
for row in results:
    print(f"{row['name']}: {row['score']}")

# 批量插入
data = [("学生" + str(i), 20, "A", 80 + i) for i in range(10)]
db.executemany("""
    INSERT INTO students (name, age, grade, score)
    VALUES (?, ?, ?, ?)
""", data)

实战示例：图书管理系统

import sqlite3
from datetime import datetime

class BookDB:
    def __init__(self, db_path="library.db"):
        self.db_path = db_path
        self.init_db()
    
    def init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS books (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    author TEXT NOT NULL,
                    isbn TEXT UNIQUE,
                    quantity INTEGER DEFAULT 1,
                    available INTEGER DEFAULT 1
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS borrowings (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    book_id INTEGER,
                    borrower TEXT,
                    borrow_date TIMESTAMP,
                    return_date TIMESTAMP,
                    FOREIGN KEY (book_id) REFERENCES books(id)
                )
            """)
    
    def add_book(self, title, author, isbn, quantity=1):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT INTO books (title, author, isbn, quantity, available)
                VALUES (?, ?, ?, ?, ?)
            """, (title, author, isbn, quantity, quantity))
    
    def borrow_book(self, book_id, borrower):
        with sqlite3.connect(self.db_path) as conn:
            book = conn.execute(
                "SELECT available FROM books WHERE id = ?", (book_id,)
            ).fetchone()
            
            if book and book[0] > 0:
                conn.execute("""
                    INSERT INTO borrowings (book_id, borrower, borrow_date)
                    VALUES (?, ?, ?)
                """, (book_id, borrower, datetime.now()))
                
                conn.execute("""
                    UPDATE books SET available = available - 1
                    WHERE id = ?
                """, (book_id,))
                return True
            return False
    
    def return_book(self, book_id):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                UPDATE borrowings 
                SET return_date = ?
                WHERE book_id = ? AND return_date IS NULL
            """, (datetime.now(), book_id))
            
            conn.execute("""
                UPDATE books SET available = available + 1
                WHERE id = ?
            """, (book_id,))

# 使用示例
db = BookDB()
db.add_book("Python编程", "张三", "978-1234567890", 5)
db.add_book("数据库原理", "李四", "978-0987654321", 3)
db.borrow_book(1, "王五")

总结

数据库编程是后端开发的核心技能。掌握SQLite后，可以轻松过渡到MySQL、PostgreSQL等更强大的数据库系统。始终使用参数化查询防止SQL注入，合理使用事务保证数据一致性，并考虑使用ORM（如SQLAlchemy）来简化复杂操作。