Python中的sqlite数据库

心本无晴.

604人浏览 · 2025-09-03 11:10:23

心本无晴. · 2025-09-03 11:10:23 发布

一、定义

SQLite 是⼀个基于 C 语⾔的轻量级 SQL 数据库引擎。它不需要独⽴的服务器进程，数据库完整

地存储在⼀个磁盘⽂件中。Python 标准库中的 sqlite3 模块提供了⼀个与 SQLite 数据库交互

的接⼝。

二、使用场景

在进⾏数据存储和管理时，我们常常需要⼀个轻量级⽽⼜功能强⼤的数据库系统。SQLite 是⼀个极受欢迎的轻量级数据库，它被⼴泛应⽤于各种⼤⼩的项⽬中。Python ⾃带了对 SQLite 的⽀持，使得在Python 中使⽤SQLite 变得⾮常简单和直观。

三、使用方法

3.1导入模块

import sqlite3

3.2创建数据库和数据库中的表

import sqlite3 # 导⼊SQLite模块
# 连接到SQLite数据库（如果数据库不存在，会⾃动创建）
conn = sqlite3.connect('students.db')
# 创建⼀个游标对象
cur = conn.cursor()
# 创建学⽣表； 创建⼀个名为 students 的表，如果不存在，则创建⼀个
"""
 id INTEGER PRIMARY KEY AUTOINCREMENT,# 整形，主键，⾃动增⻓
 name TEXT NOT NULL, # ⽂本型，不允许为空
 age INTEGER, # 整形
 grade TEXT # ⽂本型
"""
cur.execute('''
CREATE TABLE IF NOT EXISTS students (
 id INTEGER PRIMARY KEY AUTOINCREMENT,
 name TEXT NOT NULL,
 age INTEGER,
 grade TEXT
)
''')
# 提交更改并关闭连接
conn.commit() # 提交更改
conn.close() # 关闭连接

操作参考文档：SQL 教程 | 菜鸟教程https://www.runoob.com/sql/sql-tutorial.html

3.3写入数据

有了数据库和表就可以在表中写入数据

# 重新连接到数据库
conn = sqlite3.connect('students.db')
cur = conn.cursor()
# 插⼊数据
cur.execute('INSERT INTO students (name, age, grade) VALUES (?, ?, ?)', (
'Alice', 20, 'A'))
cur.execute('INSERT INTO students (name, age, grade) VALUES (?, ?, ?)', (
'Bob', 22, 'B'))
cur.execute('INSERT INTO students (name, age, grade) VALUES (?, ?, ?)', (
'Charlie', 23, 'C'))
# 提交更改并关闭连接
conn.commit()
conn.close()

3.4查询数据

conn = sqlite3.connect('students.db')
cur = conn.cursor()
# 查询数据
cur.execute('SELECT * FROM students') # 执⾏查询
rows = cur.fetchall() # 获取查询结果, 返回⼀个包含元组的列表
# 打印查询结果, 每个元组代表⼀⾏数据
for row in rows:
 print(row)
conn.close() # 关闭连接

其他语法可以查看sql教程文档

四、具体实例

存储各种汽车的Car数据库，对其进行增删改查操作。

思路：可以通过在类里构建增删改查具体方法，

第一步：导入模块，构建数据库和表

第二步：通过一个类来实例化汽车具体的属性

第三步：再次构建一个类来定义对数据库进行具体操作的方法类

第四步：调用类，实现目的

代码展示：

import sqlite3

class Car:
    def __init__(self, id, name, yanse, price, pinpai=None):
        self.id = id
        self.name = name
        self.yanse = yanse
        self.price = price
        self.pinpai = pinpai

    def __str__(self):
        return f'ID: {self.id}, 名称: {self.name}, 颜色: {self.yanse}, 价格: {self.price}, 品牌: {self.pinpai}'

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'yanse': self.yanse,
            'price': self.price,
            'pinpai': self.pinpai
        }


class CarManager:
    def __init__(self):
        self.db_name = 'cars.db'
        self.conn = sqlite3.connect(self.db_name)
        self._create_table()

    def _create_table(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS cars(
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                yanse TEXT,
                price INTEGER,
                pinpai TEXT
            )
        ''')
        self.conn.commit()

    def add_car(self, name, yanse, price, pinpai=None):
        try:
            cursor = self.conn.cursor()
            cursor.execute("SELECT id FROM cars WHERE name = ?", (name,))
            car_id = cursor.fetchone()

            if car_id is None:
                cursor.execute('INSERT INTO cars(name, yanse, price, pinpai) VALUES (?, ?, ?, ?)',
                               (name, yanse, price, pinpai))
                self.conn.commit()
                print(f'汽车{name}添加成功')
                return True
            else:
                print(f'汽车{name}已存在')
                return False
        except Exception as e:
            print(f'添加错误: {e}')
            return False

    def delete_car(self, car_id):
        try:
            cursor = self.conn.cursor()
            cursor.execute('DELETE FROM cars WHERE id = ?', (car_id,))
            rows_affected = cursor.rowcount
            self.conn.commit()
            if rows_affected > 0:
                print(f'ID为{car_id}的汽车删除成功')
                return True
            else:
                print(f'未找到ID为{car_id}的汽车')
                return False
        except Exception as e:
            print(f'删除错误: {e}')
            return False

    def update_car_by_id(self, car_id, **kwargs):
        try:
            cursor = self.conn.cursor()

            cursor.execute("SELECT id FROM cars WHERE id = ?", (car_id,))
            if cursor.fetchone() is None:
                print(f'ID为{car_id}的汽车不存在')
                return False

            if not kwargs:
                print("没有提供要更新的字段")
                return False

            valid_fields = {'name', 'yanse', 'price', 'pinpai'}
            update_fields = []
            values = []

            for field, value in kwargs.items():
                if field in valid_fields:
                    update_fields.append(f"{field} = ?")
                    values.append(value)
                else:
                    print(f"无效的字段名: {field}")
                    return False

            values.append(car_id)

            sql = f"UPDATE cars SET {', '.join(update_fields)} WHERE id = ?"
            cursor.execute(sql, values)
            self.conn.commit()

            print(f'ID为{car_id}的汽车信息更新成功')
            return True
        except Exception as e:
            print(f'更新错误: {e}')
            return False

    def get_car(self, car_id):
        try:
            cursor = self.conn.cursor()
            cursor.execute("SELECT id, name, yanse, price, pinpai FROM cars WHERE id = ?", (car_id,))
            result = cursor.fetchone()

            if result:
                return Car(result[0], result[1], result[2], result[3], result[4])
            else:
                print(f'ID为{car_id}的汽车不存在')
                return None
        except Exception as e:
            print(f'查询错误: {e}')
            return None

    def get_all_cars(self):
        try:
            cursor = self.conn.cursor()
            cursor.execute('SELECT id, name, yanse, price, pinpai FROM cars')
            results = cursor.fetchall()
            cars = []
            for row in results:
                car = Car(row[0], row[1], row[2], row[3], row[4])
                cars.append(car)
            return cars
        except Exception as e:
            print(f'查询错误: {e}')
            return []

    def display_all_cars(self):
        cars = self.get_all_cars()
        if cars:
            for car in cars:
                print(car)
        else:
            print("暂无汽车数据")

    def close(self):
        if self.conn:
            self.conn.close()


# 使用示例
if __name__ == "__main__":
    car_manager = CarManager()

    car_manager.add_car('小牛', '白色', 5000000, '兰博基尼')
    car_manager.add_car('Model S', '红色', 800000, '特斯拉')

    print("所有汽车信息:")
    car_manager.display_all_cars()

    print("\n查询ID为1的汽车:")
    car = car_manager.get_car(1)
    if car:
        print(f"汽车信息: {car}")

    print("\n修改ID为1的汽车信息:")
    car_manager.update_car_by_id(1, yanse='黑色', price=4500000)

    print("更新后的汽车信息:")
    car_manager.display_all_cars()
    car_manager.delete_car(2)
    car_manager.close()