当前位置：首页 > article >正文

Python实战：sqlite3模块应用

article 2025/1/18 6:08:59

SQLite是一个轻量级的数据库，它是一个C库，提供了一个轻量级的磁盘基数据库管理系统。Python的sqlite3模块提供了与SQLite数据库交互的功能。

一.SQLite简介

SQLite是一种嵌入式数据库，它的数据库就是一个文件。SQLite的特点如下：

不需要独立的数据库服务器进程或操作。
SQLite不需要“安装”。
SQLite事务是完全兼容ACID的，意味着它们是原子性的、一致性的、隔离性的和持久性的。
SQLite支持多数SQL92标准的功能。
SQLite是开源的，可以在商业和非商业项目中自由使用。

二.sqlite3模块安装

Python的标准库中已经包含了sqlite3模块，因此不需要额外安装。要使用sqlite3模块，只需要导入它即可：

import sqlite3

三.创建和连接数据库

使用sqlite3模块创建和连接数据库非常简单。首先，需要导入sqlite3模块，然后使用connect()函数创建数据库连接。如果数据库文件不存在，它将被自动创建。如果数据库文件已经存在，将直接连接到该数据库。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们创建了一个名为example.db的SQLite数据库文件，并创建了数据库连接。然后，我们创建了一个Cursor对象，它用于执行SQL语句。最后，我们关闭了Cursor对象和数据库连接。

四.创建表

在SQLite中，可以使用CREATE TABLE语句创建表。在Python中，可以使用sqlite3模块的execute()方法执行SQL语句。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 创建一个表
cursor.execute('''
CREATE TABLE IF NOT EXISTS user (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    age INTEGER
)
''')
# 提交事务
conn.commit()
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们创建了一个名为user的表，它有三个字段：id、name和age。id字段是主键，并且自动递增。name字段是文本类型，不允许为空。age字段是整数类型。

五.插入数据

在SQLite中，可以使用INSERT INTO语句插入数据。在Python中，可以使用sqlite3模块的execute()方法执行SQL语句。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 插入数据
cursor.execute("INSERT INTO user (name, age) VALUES (?, ?)", ('Alice', 30))
cursor.execute("INSERT INTO user (name, age) VALUES (?, ?)", ('Bob', 25))
# 提交事务
conn.commit()
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们向user表插入了两条数据。我们使用了参数化查询，将数据作为参数传递给execute()方法。这种方法可以防止SQL注入攻击。

六.查询数据

在SQLite中，可以使用SELECT语句查询数据。在Python中，可以使用sqlite3模块的execute()方法执行SQL语句，并使用fetchall()或fetchone()方法获取查询结果。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 查询数据
cursor.execute("SELECT * FROM user WHERE age >= ?", (25))
# 获取查询结果
users = cursor.fetchall()
# 打印查询结果
for user in users:
    print(user)
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们查询了年龄大于等于25岁的用户。我们使用了参数化查询，将年龄作为参数传递给execute()方法。然后，我们使用fetchall()方法获取了查询结果，并使用一个for循环打印了每个用户的详细信息。

七.更新数据

在SQLite中，可以使用UPDATE语句更新数据。在Python中，可以使用sqlite3模块的execute()方法执行SQL语句。

import sqlite3
	# 创建数据库连接
	conn = sqlite3.connect('example.db')
    # 创建一个Cursor对象，并使用它执行SQL语句
    cursor = conn.cursor()
    # 更新数据
    cursor.execute("UPDATE user SET age = ? WHERE name = ?", (28, 'Alice'))
    # 提交事务
    conn.commit()
    # 关闭Cursor对象
    cursor.close()
    # 关闭数据库连接
    conn.close()

在这个示例中，我们将名为Alice的用户的年龄更新为28岁。我们使用了参数化查询，将年龄和用户名作为参数传递给execute()方法。然后，我们提交了事务以确保更改被保存到数据库中。

八.删除数据

在SQLite中，可以使用DELETE语句删除数据。在Python中，可以使用sqlite3模块的execute()方法执行SQL语句。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 删除数据
cursor.execute("DELETE FROM user WHERE name = ?", ('Bob'))
# 提交事务
conn.commit()
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们删除了名为Bob的用户的记录。我们使用了参数化查询，将用户名作为参数传递给execute()方法。然后，我们提交了事务以确保更改被保存到数据库中。

九.使用事务

在SQLite中，可以使用事务来确保一组操作要么全部成功，要么全部失败。在Python中，可以使用sqlite3模块的commit()方法提交事务。

import sqlite3
# 创建数据库连接
conn = sqlite3.connect('example.db')
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 开始事务
conn.execute('BEGIN TRANSACTION')
try:
    # 插入数据
    cursor.execute("INSERT INTO user (name, age) VALUES (?, ?)", ('Charlie', 22))
    # 更新数据
    cursor.execute("UPDATE user SET age = ? WHERE name = ?", (32, 'Alice'))
    # 提交事务
    conn.commit()
except sqlite3.Error as e:
    # 回滚事务
    conn.rollback()
    print(f"Error: {e.args[0]}")
# 关闭Cursor对象
cursor.close()
# 关闭数据库连接
conn.close()

在这个示例中，我们首先开始一个事务，然后尝试插入一条新记录并更新一条现有记录。如果这些操作都成功，我们将提交事务。如果任何操作失败，我们将回滚事务以撤销所有更改。

十.使用SQLite数据库连接池

在高并发的应用程序中，频繁地创建和关闭数据库连接可能会导致性能问题。为了解决这个问题，可以使用数据库连接池来重用数据库连接。

import sqlite3
from sqlite3 import Connection
class Database:
    def __init__(self, db_name):
        self.conn = sqlite3.connect(db_name, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.create_tables()
    def create_tables(self):
        with self.conn as conn:
            conn.execute('''
            CREATE TABLE IF NOT EXISTS user (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                age INTEGER
            )
            ''')
    def get_connection(self):
        return self.conn
# 创建数据库实例
db = Database('example.db')
# 获取数据库连接
conn = db.get_connection()
# 创建一个Cursor对象，并使用它执行SQL语句
cursor = conn.cursor()
# 插入数据
cursor.execute("INSERT INTO user (name, age) VALUES (?, ?)", ('David', 28))
# 提交事务
conn.commit()
# 关闭Cursor对象
cursor.close()

在这个示例中，我们创建了一个Database类，它负责创建数据库连接并初始化表。我们使用check_same_thread=False参数允许在多线程环境中使用相同的连接。然后，我们创建了一个Database实例，并使用它来获取数据库连接。这样，我们就可以在整个应用程序中重用相同的连接，从而提高性能。