本文目录导读：
我来详细介绍Python操作SQLite数据库的完整方法。
基础操作
连接数据库
import sqlite3
# 连接数据库（如果不存在会自动创建）
conn = sqlite3.connect('example.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
)
''')
# 提交事务
conn.commit()
增删改查操作
# 插入数据
# 单条插入
cursor.execute("INSERT INTO students (name, age, grade, score) VALUES (?, ?, ?, ?)",
('张三', 18, '高三', 92.5))
# 多条插入
students_data = [
('李四', 17, '高二', 88.0),
('王五', 19, '高三', 95.5),
('赵六', 18, '高二', 76.0)
]
cursor.executemany("INSERT INTO students (name, age, grade, score) VALUES (?, ?, ?, ?)",
students_data)
# 查询数据
# 查询所有
cursor.execute("SELECT * FROM students")
all_students = cursor.fetchall()
for student in all_students:
print(student)
# 条件查询
cursor.execute("SELECT * FROM students WHERE score > ?", (80,))
high_score_students = cursor.fetchall()
# 更新数据
cursor.execute("UPDATE students SET score = ? WHERE name = ?", (95.0, '张三'))
# 删除数据
cursor.execute("DELETE FROM students WHERE name = ?", ('赵六',))
conn.commit()
使用上下文管理
import sqlite3
class DatabaseManager:
def __init__(self, db_name):
self.db_name = db_name
def __enter__(self):
self.conn = sqlite3.connect(self.db_name)
self.cursor = self.conn.cursor()
return self.cursor
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.conn.commit()
else:
self.conn.rollback()
self.cursor.close()
self.conn.close()
# 使用示例
with DatabaseManager('example.db') as cursor:
cursor.execute("INSERT INTO students (name, age, grade, score) VALUES (?, ?, ?, ?)",
('小红', 18, '高三', 89.0))
cursor.execute("SELECT * FROM students")
results = cursor.fetchall()
高级用法
使用字典游标
import sqlite3
conn = sqlite3.connect('example.db')
# 设置行工厂为字典类型
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM students")
rows = cursor.fetchall()
# 通过列名访问
for row in rows:
print(f"名字: {row['name']}, 分数: {row['score']}")
批量导入数据
import csv
import sqlite3
def import_from_csv(csv_file, db_name):
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
cursor.execute(
"INSERT INTO students (name, age, grade, score) VALUES (?, ?, ?, ?)",
(row['name'], int(row['age']), row['grade'], float(row['score']))
)
conn.commit()
conn.close()
错误处理
import sqlite3
from sqlite3 import Error
def safe_database_operation():
try:
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
# 执行操作
cursor.execute("SELECT * FROM students")
return cursor.fetchall()
except Error as e:
print(f"数据库错误: {e}")
return None
except Exception as e:
print(f"其他错误: {e}")
return None
finally:
if conn:
conn.close()
完整示例：学生管理系统
import sqlite3
from sqlite3 import Error
class StudentManager:
def __init__(self, db_name='students.db'):
self.db_name = db_name
self.conn = None
self.init_database()
def init_database(self):
"""初始化数据库和表"""
try:
self.conn = sqlite3.connect(self.db_name)
cursor = self.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
)
''')
self.conn.commit()
except Error as e:
print(f"初始化数据库失败: {e}")
def add_student(self, name, age, grade, score):
"""添加学生"""
try:
cursor = self.conn.cursor()
cursor.execute(
"INSERT INTO students (name, age, grade, score) VALUES (?, ?, ?, ?)",
(name, age, grade, score)
)
self.conn.commit()
return cursor.lastrowid
except Error as e:
print(f"添加学生失败: {e}")
return None
def get_all_students(self):
"""获取所有学生"""
try:
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM students ORDER BY score DESC")
return cursor.fetchall()
except Error as e:
print(f"查询失败: {e}")
return []
def search_students(self, keyword):
"""搜索学生"""
try:
cursor = self.conn.cursor()
cursor.execute(
"SELECT * FROM students WHERE name LIKE ?",
(f'%{keyword}%',)
)
return cursor.fetchall()
except Error as e:
print(f"搜索失败: {e}")
return []
def update_score(self, student_id, new_score):
"""更新分数"""
try:
cursor = self.conn.cursor()
cursor.execute(
"UPDATE students SET score = ? WHERE id = ?",
(new_score, student_id)
)
self.conn.commit()
return cursor.rowcount > 0
except Error as e:
print(f"更新失败: {e}")
return False
def delete_student(self, student_id):
"""删除学生"""
try:
cursor = self.conn.cursor()
cursor.execute("DELETE FROM students WHERE id = ?", (student_id,))
self.conn.commit()
return cursor.rowcount > 0
except Error as e:
print(f"删除失败: {e}")
return False
def close(self):
"""关闭连接"""
if self.conn:
self.conn.close()
# 使用示例
if __name__ == "__main__":
manager = StudentManager()
# 添加学生
manager.add_student('张三', 18, '高三', 92.5)
manager.add_student('李四', 17, '高二', 88.0)
# 查询所有学生
print("所有学生:")
for student in manager.get_all_students():
print(student)
# 搜索学生
print("\n搜索'张':")
for student in manager.search_students('张'):
print(student)
# 关闭连接
manager.close()
常用函数
# 获取表名
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
# 获取表结构
cursor.execute("PRAGMA table_info(students)")
columns = cursor.fetchall()
# 获取当前数据库大小
import os
db_size = os.path.getsize('example.db')
# 执行SQL脚本文件
with open('script.sql', 'r', encoding='utf-8') as f:
script = f.read()
cursor.executescript(script)
注意事项
- 防SQL注入：始终使用参数化查询（占位符）
- 事务管理：批量操作时使用事务提高性能
- 连接管理：使用完及时关闭连接
- 编码问题：指定UTF-8编码处理中文
这些就是Python操作SQLite的基本方法,可以根据实际需求选择合适的方案。
