Python中如何使用sqlite3操作SQLite数据库详解

import sqlite3

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

print("数据库连接成功")

cursor = conn.cursor()

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        city TEXT
    )
''')
conn.commit()  # 提交更改
print("表创建成功")

cursor.execute("INSERT INTO users (name, age, city) VALUES (?, ?, ?)",
               ("Alice", 25, "New York"))
conn.commit()
print("插入数据成功")

users = [
    ("Bob", 30, "San Francisco"),
    ("Charlie", 28, "Los Angeles"),
    ("David", 35, "Seattle")
]

cursor.executemany("INSERT INTO users (name, age, city) VALUES (?, ?, ?)", users)
conn.commit()
print("批量插入成功")

cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()

for row in rows:
    print(row)

(1, 'Alice', 25, 'New York')
(2, 'Bob', 30, 'San Francisco')
(3, 'Charlie', 28, 'Los Angeles')

cursor.execute("SELECT * FROM users WHERE name = ?", ("Alice",))
user = cursor.fetchone()
print(user)

cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, "Alice"))
conn.commit()
print("数据更新成功")

cursor.execute("DELETE FROM users WHERE name = ?", ("David",))
conn.commit()
print("数据删除成功")

conn.commit()  # 确保数据写入数据库

try:
    cursor.execute("UPDATE users SET age = ? WHERE name = ?", (100, "Charlie"))
    raise Exception("模拟异常")  # 模拟错误
    conn.commit()
except Exception as e:
    conn.rollback()  # 发生错误时回滚
    print("事务回滚:", e)

cursor.close()
conn.close()
print("数据库连接已关闭")

with sqlite3.connect("example.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

cursor.execute("SELECT * FROM users")
print(cursor.fetchone())  # (1, 'Alice', 25, 'New York')

conn.row_factory = sqlite3.Row  # 设置行工厂
cursor = conn.cursor()

cursor.execute("SELECT * FROM users")
row = cursor.fetchone()
print(row["name"], row["age"])  # Alice 25