Python高级技术之：`Python`的`SQLite`：在`unittest`和小型应用中的使用。

嘿，各位，今天咱们来聊聊Python里那个“小而美”的数据库——SQLite，以及它在unittest和小型应用中的妙用。

先声明一下，我不是那种会跟你吹嘘“五分钟精通SQLite”的速成班讲师，咱们要讲的是实实在在能用上的东西。准备好了吗？那我们就开始吧！

SQLite：数据库界的瑞士军刀

如果把数据库比作工具箱，那SQLite绝对是那把瑞士军刀。它体积小巧，无需单独的服务器进程，直接读写文件，这意味着啥？意味着部署超简单！想想你用MySQL或者PostgreSQL的时候，是不是得先装个服务器，配置一堆东西？SQLite就省了这些麻烦。

它虽然小，功能可一点都不弱。支持标准的SQL语法，事务，索引，触发器等等，该有的都有。特别适合小型项目，测试环境，或者嵌入式设备。

Python与SQLite：天生一对

Python自带了sqlite3模块，让你能轻松地操作SQLite数据库。这简直是程序员的福音！不用额外安装任何东西，直接就能用。

import sqlite3

# 连接到数据库 (如果文件不存在，会自动创建)
conn = sqlite3.connect('my_database.db')

# 创建一个游标对象
cursor = conn.cursor()

# 创建一个表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        age INTEGER
    )
''')

# 插入数据
cursor.execute("INSERT INTO users (name, age) VALUES ('Alice', 30)")
cursor.execute("INSERT INTO users (name, age) VALUES ('Bob', 25)")

# 提交更改
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()

for row in rows:
    print(row)

# 关闭连接
conn.close()

这段代码演示了SQLite的基本操作：连接数据库，创建表，插入数据，查询数据，最后关闭连接。是不是感觉很简单？

在unittest中使用SQLite：测试界的利器

在单元测试中，使用真实的数据库往往会带来一些问题：

• 依赖性： 测试依赖于外部数据库，环境一旦改变，测试就可能失败。
• 速度： 真实的数据库操作速度通常较慢，会拖慢测试速度。
• 数据污染： 测试过程中可能会修改数据库中的数据，影响其他测试或者生产环境。

SQLite可以完美解决这些问题。我们可以为每个测试创建一个临时的SQLite数据库，测试结束后自动销毁。这样，测试就是完全隔离的，速度也很快。

import unittest
import sqlite3

class TestUserDatabase(unittest.TestCase):

    def setUp(self):
        # 创建一个内存数据库
        self.conn = sqlite3.connect(':memory:') # 使用内存数据库，测试结束后自动销毁
        self.cursor = self.conn.cursor()

        # 创建表
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                age INTEGER
            )
        ''')
        self.conn.commit()

    def tearDown(self):
        # 关闭连接
        self.conn.close()

    def test_insert_user(self):
        # 插入一个用户
        self.cursor.execute("INSERT INTO users (name, age) VALUES ('Charlie', 35)")
        self.conn.commit()

        # 查询用户
        self.cursor.execute("SELECT * FROM users WHERE name = 'Charlie'")
        user = self.cursor.fetchone()

        # 断言
        self.assertEqual(user[1], 'Charlie')
        self.assertEqual(user[2], 35)

    def test_get_user_by_id(self):
        # 插入一个用户
        self.cursor.execute("INSERT INTO users (name, age) VALUES ('David', 40)")
        self.conn.commit()
        self.cursor.execute("SELECT id FROM users WHERE name = 'David'")
        user_id = self.cursor.fetchone()[0]

        # 查询用户
        self.cursor.execute(f"SELECT * FROM users WHERE id = {user_id}") # Corrected SQL injection vulnerability
        user = self.cursor.fetchone()

        # 断言
        self.assertEqual(user[1], 'David')
        self.assertEqual(user[2], 40)

if __name__ == '__main__':
    unittest.main()

在这个例子中，setUp方法负责创建内存数据库和表，tearDown方法负责关闭连接。每个测试方法都可以在独立的数据库环境中运行，互不影响。

注意：conn = sqlite3.connect(':memory:') 这行代码创建了一个内存数据库。 内存数据库在连接关闭后会自动销毁，非常适合用于测试。

小型应用中的SQLite：麻雀虽小，五脏俱全

SQLite非常适合用于开发小型应用，比如：

• 桌面应用： 存储用户配置，缓存数据。
• 移动应用： 存储本地数据，比如联系人，日历事件。
• 脚本： 存储脚本运行过程中产生的数据。

下面是一个简单的例子，演示如何使用SQLite存储待办事项列表：

import sqlite3

def create_table():
    conn = sqlite3.connect('todo.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS todos (
            id INTEGER PRIMARY KEY,
            task TEXT NOT NULL,
            completed INTEGER DEFAULT 0
        )
    ''')
    conn.commit()
    conn.close()

def add_task(task):
    conn = sqlite3.connect('todo.db')
    cursor = conn.cursor()
    cursor.execute("INSERT INTO todos (task) VALUES (?)", (task,)) # 使用参数化查询，防止SQL注入
    conn.commit()
    conn.close()

def mark_completed(task_id):
    conn = sqlite3.connect('todo.db')
    cursor = conn.cursor()
    cursor.execute("UPDATE todos SET completed = 1 WHERE id = ?", (task_id,)) # 使用参数化查询，防止SQL注入
    conn.commit()
    conn.close()

def list_tasks():
    conn = sqlite3.connect('todo.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM todos")
    tasks = cursor.fetchall()
    conn.close()
    return tasks

def main():
    create_table()

    while True:
        print("nTodo List:")
        tasks = list_tasks()
        for task in tasks:
            print(f"{task[0]}. [{ 'x' if task[2] else ' ' }] {task[1]}")

        print("nOptions:")
        print("1. Add task")
        print("2. Mark task as completed")
        print("3. Exit")

        choice = input("Enter your choice: ")

        if choice == '1':
            task = input("Enter task: ")
            add_task(task)
        elif choice == '2':
            task_id = input("Enter task ID to mark as completed: ")
            mark_completed(task_id)
        elif choice == '3':
            break
        else:
            print("Invalid choice.")

if __name__ == '__main__':
    main()

这个程序可以添加待办事项，标记完成状态，并列出所有待办事项。代码虽然简单，但是已经具备了一个基本应用的功能。

SQLite的优势总结

• 轻量级： 体积小巧，资源占用少。
• 无需服务器： 直接读写文件，部署简单。
• 易于使用： Python自带sqlite3模块，操作方便。
• 跨平台： 支持多种操作系统。
• ACID事务： 保证数据完整性。

SQLite的局限性

• 并发性能： 不适合高并发的场景。
• 数据量： 不适合存储海量数据。
• 网络访问： 不支持网络访问，只能在本地使用。

SQLite的适用场景

场景	是否适用	说明
小型应用	适用	存储用户配置，缓存数据等。
单元测试	适用	提供隔离的测试环境，加速测试速度。
嵌入式设备	适用	资源有限，需要轻量级数据库。
高并发应用	不适用	并发性能有限，容易出现锁竞争。
海量数据存储	不适用	性能会受到影响，建议使用更专业的数据库。
需要网络访问的应用	不适用	SQLite只能在本地使用。

防止SQL注入

在使用SQLite时，一定要注意防止SQL注入攻击。

错误示范：

task_id = input("Enter task ID: ")
cursor.execute(f"SELECT * FROM todos WHERE id = {task_id}") # 存在SQL注入风险！

如果用户输入 1 OR 1=1，那么SQL语句就会变成 SELECT * FROM todos WHERE id = 1 OR 1=1，这会导致所有数据都被查询出来。

正确示范：

task_id = input("Enter task ID: ")
cursor.execute("SELECT * FROM todos WHERE id = ?", (task_id,)) # 使用参数化查询，安全！

使用参数化查询，可以有效地防止SQL注入攻击。参数化查询会将用户输入的数据作为参数传递给SQL语句，而不是直接拼接SQL语句。这样，即使用户输入恶意代码，也不会被当做SQL语句执行。

总结

SQLite是一个非常优秀的轻量级数据库，特别适合用于unittest和小型应用。它简单易用，功能强大，可以满足很多场景的需求。当然，它也有一些局限性，需要根据实际情况选择合适的数据库。希望今天的讲解能让你对SQLite有更深入的了解。

好了，今天的讲座就到这里。有问题可以随时提问。下次有机会，我们再聊聊其他有趣的技术话题。拜拜！