喜欢的条友记得关注、点赞、转发、收藏，你们的支持就是我最大的动力源泉。

前期基础教程：

「Python3.11.0」手把手教你安装最新版Python运行环境

讲讲Python环境使用Pip命令快速下载各类库的方法

Python启航：30天编程速成之旅（第2天）-IDE安装

【Python教程】JupyterLab 开发环境安装

一、设计理念与核心思路

今天突发奇想，想实现自己的数据库引擎，通过JSON文件模拟数据库文件，用面向对象思想实现SQLite3的五大核心功能：

1. 极简存储：采用JSON文件替代传统数据库文件，单文件存储所有数据（类似SQLite的.db文件特性）
2. 表结构管理：通过字典嵌套实现表结构存储，包含列定义和数据行
3. CRUD操作：完整实现Create(创建)、Read(读取)、Update(更新)、Delete(删除)四大操作
4. 条件查询：基于列索引的条件过滤机制
5. 数据持久化：每次操作自动保存到JSON文件（类似SQLite的事务提交）

与传统SQLite3相比的差异化设计：

• 使用列表索引代替SQL语句
• JSON可视化存储替代二进制存储
• 轻量化实现（无需安装数据库引擎）

二、核心功能实现代码解析

（1）数据库初始化

def __init__(self, db_file):
    self.db_file = db_file
    try:
        with open(self.db_file, 'r') as f:
            self.data = json.load(f)
    except FileNotFoundError:
        self.data = {}

• 自动检测并加载现有数据库文件
• 文件不存在时创建空数据库（类似SQLite的自动创建特性）
• 使用字典结构存储整个数据库（顶层键为表名）

（2）表结构创建

def create_table(self, table_name, columns):
    if table_name not in self.data:
        self.data[table_name] = {
            "columns": columns,  # 存储列定义
            "rows": []           # 数据行容器
        }

• 实现CREATE TABLE等价功能
• 支持列名定义（类似SQLite的schema）
• 自动检测表名冲突（防重复创建）

（3）数据插入

def insert(self, table_name, values):
    if len(values) == len(table["columns"]):
        table["rows"].append(values)

• 实现INSERT INTO等价功能
• 列值数量验证机制
• 支持单行插入（可通过循环实现批量插入）

（4）条件查询

def select(self, table_name, where=None):
    if where is None:
        return table["rows"]
    else:
        return [row for row in table["rows"] if row[column_index] == value]

• 实现基础版SELECT查询
• 支持全表扫描和条件过滤
• 使用列表推导式实现WHERE子句功能

（5）数据更新

def update(...):
    for row in table["rows"]:
        if row[column_index] == value:
            for i, val in enumerate(set_values):
                if val is not None:
                    row[i] = val

• 实现UPDATE等价功能
• 支持多列同时更新
• None值保留原始数据的特性

（6）数据删除

table["rows"] = [row for row in table["rows"] 
                if row[column_index] != value]

• 实现DELETE FROM等价功能
• 通过列表推导式过滤目标数据
• 支持单条件删除

三、与SQLite3的对比分析

四、下一步优化内容

1. 增加事务批处理：将多个操作打包提交
2. 实现索引功能：为常用查询字段建立哈希索引
3. 支持复杂查询：添加>、<、LIKE等条件运算符
4. 数据类型验证：根据列定义校验输入类型
5. 连接池管理：支持多线程访问
6. 数据备份机制：定期生成数据库快照

五、应用场景推荐

1. 个人配置存储：用户设置、收藏夹等小数据量场景
2. 原型开发测试：快速验证业务逻辑的临时数据库
3. 教育演示工具：直观展示数据库工作原理
4. IoT设备存储：资源受限设备的轻量级存储方案

六、完整使用示例

import json

class SimpleDatabase:
    def __init__(self, db_file):
        self.db_file = db_file
        try:
            with open(self.db_file, 'r') as f:
                self.data = json.load(f)
        except FileNotFoundError:
            self.data = {}

    def save(self):
        with open(self.db_file, 'w') as f:
            json.dump(self.data, f, indent=4)

    def create_table(self, table_name, columns):
        """
        创建一个新的表
        :param table_name: 表名
        :param columns: 列名列表
        """
        if table_name not in self.data:
            self.data[table_name] = {
                "columns": columns,
                "rows": []
            }
            self.save()
        else:
            print(f"Table {table_name} already exists.")

    def insert(self, table_name, values):
        """
        向指定表中插入一行数据
        :param table_name: 表名
        :param values: 要插入的值列表
        """
        if table_name in self.data:
            table = self.data[table_name]
            if len(values) == len(table["columns"]):
                table["rows"].append(values)
                self.save()
            else:
                print("Number of values does not match number of columns.")
        else:
            print(f"Table {table_name} does not exist.")

    def select(self, table_name, where=None):
        """
        从指定表中查询数据
        :param table_name: 表名
        :param where: 查询条件，格式为 (column_index, value)
        :return: 查询结果列表
        """
        if table_name in self.data:
            table = self.data[table_name]
            if where is None:
                return table["rows"]
            else:
                column_index, value = where
                return [row for row in table["rows"] if row[column_index] == value]
        else:
            print(f"Table {table_name} does not exist.")
            return []

    def update(self, table_name, set_values, where):
        """
        更新指定表中的数据
        :param table_name: 表名
        :param set_values: 要更新的值列表
        :param where: 更新条件，格式为 (column_index, value)
        """
        if table_name in self.data:
            table = self.data[table_name]
            column_index, value = where
            for row in table["rows"]:
                if row[column_index] == value:
                    for i, val in enumerate(set_values):
                        if val is not None:
                            row[i] = val
            self.save()
        else:
            print(f"Table {table_name} does not exist.")

    def delete(self, table_name, where):
        """
        从指定表中删除数据
        :param table_name: 表名
        :param where: 删除条件，格式为 (column_index, value)
        """
        if table_name in self.data:
            table = self.data[table_name]
            column_index, value = where
            table["rows"] = [row for row in table["rows"] if row[column_index] != value]
            self.save()
        else:
            print(f"Table {table_name} does not exist.")


# 使用示例
if __name__ == "__main__":
    db = SimpleDatabase('test_db.json')

    # 创建表
    db.create_table('users', ['id', 'name', 'age'])

    # 插入数据
    db.insert('users', [1, 'Alice', 25])
    db.insert('users', [2, 'Bob', 30])

    # 查询数据
    results = db.select('users')
    print("All users:", results)

    # 更新数据
    db.update('users', [None, 'Alice Smith', None], (0, 1))
    results = db.select('users')
    print("Users after update:", results)

    # 删除数据
    db.delete('users', (0, 2))
    results = db.select('users')
    print("Users after delete:", results)

运行结果（我已经建立了user表，所以提示：Table users already exists.）：

数据库文件test_db.json内容：

通过这个实现，条友可以深入理解数据库引擎的核心原理。建议结合官方sqlite3模块（import sqlite3）对比学习，体验完整数据库功能

。本代码已开源，关注作者获取最新优化版本。

喜欢的条友记得关注、点赞、转发、收藏，你们的支持就是我最大的动力源泉。