Python脚本如何操作SQLite数据库

本文目录导读：

1. 基础操作
2. 使用上下文管理
3. 高级用法
4. 完整示例：学生管理系统
5. 常用函数
6. 注意事项

我来详细介绍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)

注意事项

1. 防SQL注入：始终使用参数化查询（占位符）
2. 事务管理：批量操作时使用事务提高性能
3. 连接管理：使用完及时关闭连接
4. 编码问题：指定UTF-8编码处理中文

这些就是Python操作SQLite的基本方法,可以根据实际需求选择合适的方案。