怎样用Python操作SQLite？轻量数据库使用指南

import sqlite3

# 连接到数据库（如果不存在则创建）
# 也可以使用 ':memory:' 在内存中创建一个临时数据库，关闭连接后数据即消失
conn = sqlite3.connect('my_database.db')

# 创建一个游标对象
cursor = conn.cursor()

# 执行SQL命令：创建表
# 这里我喜欢用三重引号，写多行SQL看着舒服
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
''')

# 插入数据
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ('Alice', 'alice@example.com'))
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ('Bob', 'bob@example.com'))
# 批量插入是个好习惯，效率高
users_to_add = [
    ('Charlie', 'charlie@example.com'),
    ('David', 'david@example.com')
]
cursor.executemany("INSERT INTO users (name, email) VALUES (?, ?)", users_to_add)


# 查询数据
cursor.execute("SELECT * FROM users WHERE name LIKE ?", ('%ice%',))
results = cursor.fetchall() # 获取所有匹配的行
print("查询结果 (Alice):", results)

# 更新数据
cursor.execute("UPDATE users SET email = ? WHERE name = ?", ('alice.updated@example.com', 'Alice'))

# 删除数据
cursor.execute("DELETE FROM users WHERE name = ?", ('Bob',))

# 提交更改。这步非常关键，没有它，你的数据操作不会真正保存到数据库文件
conn.commit()
print("数据已提交。")

# 再次查询所有数据，看看变化
cursor.execute("SELECT * FROM users")
all_users = cursor.fetchall()
print("所有用户:", all_users)

# 关闭游标和连接，释放资源
cursor.close()
conn.close()
print("数据库连接已关闭。")

import sqlite3

# 连接到文件数据库
file_db_conn = sqlite3.connect('my_app_data.db')
print("已连接到文件数据库 'my_app_data.db'")

# 连接到内存数据库
memory_db_conn = sqlite3.connect(':memory:')
print("已连接到内存数据库 (临时)")

# 通常，我会用try-finally或者with语句来管理连接，确保连接总是被关闭
# 这种方式更健壮，即使代码执行过程中出现错误，也能保证资源释放
try:
    cursor = file_db_conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT)")
    file_db_conn.commit()
    print("文件数据库表已创建。")
except Exception as e:
    print(f"连接或操作文件数据库时出错: {e}")
finally:
    file_db_conn.close()
    print("文件数据库连接已关闭。")

# 内存数据库也是一样
try:
    cursor_mem = memory_db_conn.cursor()
    cursor_mem.execute("CREATE TABLE IF NOT EXISTS temp_data (id INTEGER, info TEXT)")
    cursor_mem.execute("INSERT INTO temp_data VALUES (1, '临时数据')")
    memory_db_conn.commit()
    cursor_mem.execute("SELECT * FROM temp_data")
    print("内存数据库数据:", cursor_mem.fetchall())
except Exception as e:
    print(f"连接或操作内存数据库时出错: {e}")
finally:
    memory_db_conn.close()
    print("内存数据库连接已关闭。")

import sqlite3

# 使用with语句管理连接，更安全、更简洁
with sqlite3.connect('another_db.db') as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT)")
    cursor.execute("INSERT INTO products (name) VALUES ('Laptop')")
    conn.commit() # 事务管理依然需要手动commit
    print("使用with语句：产品表已创建并插入数据。")

# 连接在with块结束后会自动关闭

# 插入单条数据
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ('Frank', 'frank@example.com'))

# 插入多条数据，使用executemany效率更高
new_users = [
    ('Grace', 'grace@example.com'),
    ('Heidi', 'heidi@example.com')
]
cursor.executemany("INSERT INTO users (name, email) VALUES (?, ?)", new_users)

# 查询所有用户
cursor.execute("SELECT id, name, email FROM users")
all_users = cursor.fetchall()
print("所有用户:", all_users)

# 查询特定用户
cursor.execute("SELECT * FROM users WHERE name = ?", ('Grace',))
grace_info = cursor.fetchone() # 即使只有一条，也用fetchone
print("Grace的信息:", grace_info)

# 遍历查询结果
for row in cursor.execute("SELECT name FROM users"):
    print(f"用户名字: {row[0]}") # row是一个元组

cursor.execute("UPDATE users SET email = ? WHERE name = ?", ('frank.new@example.com', 'Frank'))

cursor.execute("DELETE FROM users WHERE name = ?", ('Heidi',))

conn = sqlite3.connect('transaction_test.db')
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS accounts (id INTEGER PRIMARY KEY, name TEXT, balance REAL)")
conn.commit() # 确保表已创建

try:
    cursor.execute("INSERT INTO accounts (name, balance) VALUES (?, ?)", ('AccountA', 1000.0))
    cursor.execute("INSERT INTO accounts (name, balance) VALUES (?, ?)", ('AccountB', 500.0))

    # 模拟转账操作：从A扣钱，给B加钱
    cursor.execute("UPDATE accounts SET balance = balance - 200 WHERE name = ?", ('AccountA',))
    # 假设这里发生了一个错误，比如网络中断或者其他异常
    # raise ValueError("模拟一个错误")
    cursor.execute("UPDATE accounts SET balance = balance + 200 WHERE name = ?", ('AccountB',))

    conn.commit() # 只有这里执行了，数据才会真正保存
    print("转账成功，事务提交。")
except Exception as e:
    conn.rollback() # 出现错误时回滚所有操作
    print(f"转账失败，事务回滚: {e}")
finally:
    cursor.close()
    conn.close()

# 检查结果
with sqlite3.connect('transaction_test.db') as conn_check:
    cursor_check = conn_check.cursor()
    cursor_check.execute("SELECT * FROM accounts")
    print("当前账户余额:", cursor_check.fetchall())

# 安全的做法：使用占位符 '?'
user_name = "Robert'; DROP TABLE users;" # 恶意输入
cursor.execute("SELECT * FROM users WHERE name = ?", (user_name,))
# SQLite会将 'Robert'; DROP TABLE users; 作为一个整体字符串来匹配，不会执行DROP TABLE

# 批量插入示例，性能优于单条循环插入
data_to_insert = [(f'User{i}', f'user{i}@example.com') for i in range(10000)]
conn = sqlite3.connect('bulk_insert.db')
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS large_table (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")

# 在一个事务中完成所有插入
try:
    cursor.executemany("INSERT INTO large_table (name, email) VALUES (?, ?)", data_to_insert)
    conn.commit()
    print("10000条数据批量插入完成。")
except Exception as e:
    conn.rollback()
    print(f"批量插入失败: {e}")
finally:
    conn.close()

CREATE INDEX idx_users_email ON users (email);
CREATE INDEX idx_products_name ON products (name);

# 谨慎使用，可能导致数据丢失
cursor.execute("PRAGMA synchronous = OFF;")
conn.commit() # 提交PRAGMA设置

# 执行VACUUM操作
conn.execute("VACUUM;")
conn.commit()
print("数据库已VACUUM。")