SQLite - Python

SQLite 与 Python

Python 通过 sqlite3 模块内置了对 SQLite 的支持，该模块是标准库的一部分。这使得将功能强大的文件型数据库集成到 Python 应用程序中变得异常简单，无需外部依赖。本教程涵盖了使用 sqlite3 的现代最佳实践。

核心概念

sqlite3 模块提供了符合 DB-API 2.0 规范（PEP 249）的 SQL 接口。您将主要使用的对象是：

• 连接对象 (Connection Object)：表示数据库连接。用于管理事务（commit 提交，rollback 回滚）和创建 Cursor 对象。
• 游标对象 (Cursor Object)：用于执行 SQL 查询、获取结果和检查查询元数据。

连接到数据库

管理连接的最佳实践是使用 with 语句，即使发生错误，它也会自动关闭连接。

# 现代 Python 3 语法
import sqlite3

db_file = 'company_database.db'

try:
    # 'with' 语句确保连接自动关闭
    with sqlite3.connect(db_file) as conn:
        print(f"Successfully connected to database {db_file}")
        # 现在您可以创建一个游标并执行查询
        # cursor = conn.cursor()
except sqlite3.Error as e:
    print(f"Database error: {e}")

# 要使用内存数据库，请传入 ":memory:"
# with sqlite3.connect(':memory:') as conn:

执行查询：最佳实践

始终使用参数化查询（带 ? 占位符）以防止 SQL 注入漏洞。切勿使用 f-string 或字符串格式化将变量数据插入到您的查询中。

CREATE TABLE (创建表)

import sqlite3

# 为连接和游标使用上下文管理器
def create_table(conn):
    try:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS employees (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                department TEXT NOT NULL,
                salary REAL
            );
        ''')
        print("Table 'employees' created or already exists.")
    except sqlite3.Error as e:
        print(f"Error creating table: {e}")

with sqlite3.connect('company_database.db') as conn:
    create_table(conn)

INSERT 操作 (插入数据)

要插入单条记录，请使用 cursor.execute()。对于多条记录，cursor.executemany() 的效率要高得多。

import sqlite3

def add_employee(conn, name, department, salary):
    sql = 'INSERT INTO employees(name, department, salary) VALUES(?, ?, ?)'
    try:
        cursor = conn.cursor()
        cursor.execute(sql, (name, department, salary))
        conn.commit() # 提交事务以保存更改
        print(f"Added employee: {name}")
    except sqlite3.Error as e:
        print(f"Error adding employee: {e}")
        conn.rollback() # 出错时回滚

def add_multiple_employees(conn, employees_list):
    sql = 'INSERT INTO employees(name, department, salary) VALUES(?, ?, ?)'
    try:
        cursor = conn.cursor()
        cursor.executemany(sql, employees_list)
        conn.commit()
        print(f"Added {len(employees_list)} new employees.")
    except sqlite3.Error as e:
        print(f"Error adding multiple employees: {e}")
        conn.rollback()

with sqlite3.connect('company_database.db') as conn:
    # 单条插入
    add_employee(conn, 'Alice', 'Engineering', 80000.00)

    # 多条插入
    new_staff = [
        ('Bob', 'Marketing', 65000.00),
        ('Charlie', 'Engineering', 95000.00),
        ('Diana', 'HR', 72000.00)
    ]
    add_multiple_employees(conn, new_staff)

SELECT 操作 (查询数据)

为了让结果更容易处理，您可以将连接的 row_factory 设置为 sqlite3.Row。这允许您像字典一样通过名称访问列。

import sqlite3

def get_engineers(conn):
    # 设置 row_factory 以通过名称访问列
    conn.row_factory = sqlite3.Row

    sql = 'SELECT * FROM employees WHERE department = ? ORDER BY name'
    try:
        cursor = conn.cursor()
        cursor.execute(sql, ('Engineering',))

        engineers = cursor.fetchall() # 使用 fetchall()、fetchone() 或 fetchmany()

        print("\n--- Engineering Team ---")
        for emp in engineers:
            print(f"ID: {emp['id']}, Name: {emp['name']}, Salary: ${emp['salary']:.2f}")

    except sqlite3.Error as e:
        print(f"Error fetching data: {e}")

with sqlite3.connect('company_database.db') as conn:
    get_engineers(conn)

UPDATE 操作 (更新数据)

import sqlite3

def give_raise(conn, employee_id, new_salary):
    sql = 'UPDATE employees SET salary = ? WHERE id = ?'
    try:
        cursor = conn.cursor()
        cursor.execute(sql, (new_salary, employee_id))
        conn.commit()
        if cursor.rowcount == 0:
            print(f"No employee found with ID {employee_id}.")
        else:
            print(f"Updated salary for employee {employee_id}. Total changes: {conn.total_changes}")
    except sqlite3.Error as e:
        print(f"Error updating data: {e}")
        conn.rollback()

with sqlite3.connect('company_database.db') as conn:
    give_raise(conn, 1, 85000.00) # 给 Alice 加薪

DELETE 操作 (删除数据)

import sqlite3

def delete_employee(conn, employee_id):
    sql = 'DELETE FROM employees WHERE id = ?'
    try:
        cursor = conn.cursor()
        cursor.execute(sql, (employee_id,))
        conn.commit()
        if cursor.rowcount == 0:
            print(f"No employee found with ID {employee_id}.")
        else:
            print(f"Deleted employee with ID {employee_id}.")
    except sqlite3.Error as e:
        print(f"Error deleting data: {e}")
        conn.rollback()

with sqlite3.connect('company_database.db') as conn:
    delete_employee(conn, 2) # 删除 Bob

进一步学习：ORM (对象关系映射)

对于大型应用程序，编写原始 SQL 可能会变得繁琐。对象关系映射器 (ORM) 允许您使用 Python 对象与数据库进行交互。它们可以提高生产力并减少错误。考虑探索流行的库，例如：

• SQLAlchemy：一个功能强大且全面的 ORM 和 SQL 工具包，适用于大型应用程序。
• Peewee：一个简单、小巧且富有表现力的 ORM，非常适合小型项目。