介绍
在这个实验中，你将学习如何在 SQLite 中处理 JSON 数据。你将探索如何在 SQLite 数据库中存储、提取、过滤和更新 JSON 数据。这个实验提供了一个关于在 SQLite 中使用 JSON 数据的实践性介绍，这项技能在现代数据管理中变得越来越有价值。
创建数据库和表
在这个步骤中，你将创建一个 SQLite 数据库和一个用于存储 JSON 数据的表。SQLite 是一个轻量级数据库，它将数据存储在单个文件中，从而易于管理。
首先，打开你的终端。默认路径是 /home/labex/project。
现在，让我们创建一个目录来存储我们的数据库。
mkdir sqlite_json
cd sqlite_json
这些命令创建了一个名为 sqlite_json 的目录，然后将当前目录更改为它。这将使你的项目文件保持井井有条。
接下来，创建一个名为 mydatabase.db 的 SQLite 数据库。
sqlite3 mydatabase.db
这个命令打开 SQLite shell，连接到 mydatabase.db 数据库。如果数据库文件不存在，SQLite 将创建它。
现在，创建一个名为 products 的表，其中包含 id、name 和 details 列。details 列将以文本形式存储 JSON 数据。
CREATE TABLE products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT,
details TEXT
);
这个 SQL 命令创建了 products 表：
id：一个唯一的整数，对于每个新产品自动递增。name：产品的名称（例如，Laptop，Smartphone）。details：一个文本字段，用于存储产品的 JSON 数据。
插入 JSON 数据
在这个步骤中，你将把 JSON 数据插入到 products 表中。
让我们将两个示例记录插入到 products 表中。
INSERT INTO products (name, details) VALUES (
'Laptop',
'{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}'
);
INSERT INTO products (name, details) VALUES (
'Smartphone',
'{"brand": "Samsung", "model": "Galaxy S21", "specs": {"display": "6.2 inch", "camera": "12MP", "storage": "128GB"}}'
);
这些 INSERT 语句向 products 表添加了两行数据。details 列包含作为文本字符串的 JSON 数据。
为了验证数据是否已正确插入，请运行以下查询：
SELECT * FROM products;
预期输出：
1|Laptop|{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}
2|Smartphone|{"brand": "Samsung", "model": "Galaxy S21", "specs": {"display": "6.2 inch", "camera": "12MP", "storage": "128GB"}}
此输出确认 JSON 数据已成功存储在 products 表中。
使用自定义函数提取 JSON 字段
由于 SQLite 没有内置的 JSON 函数，你将创建一个自定义 Python 函数来从 JSON 字符串中提取数据。
首先，退出 SQLite shell。
.exit
现在，创建一个名为 json_extractor.py 的 Python 文件。
nano json_extractor.py
将以下 Python 代码粘贴到 json_extractor.py 文件中：
## 导入必要的库
import sqlite3
import json
## 定义一个函数，使用路径从 JSON 字符串中提取值
def json_extract(json_str, path):
try:
## 将 JSON 字符串解析为 Python 字典
json_data = json.loads(json_str)
## 将路径拆分为组件（例如，'specs.cpu' 变为 ['specs', 'cpu']）
path_components = path.split('.')
## 从完整的 JSON 对象开始
value = json_data
## 使用路径组件遍历 JSON 对象
for component in path_components:
## 获取当前组件的值
value = value.get(component)
## 返回最终值
return value
## 处理 JSON 无效或路径不存在的错误
except (json.JSONDecodeError, AttributeError, TypeError):
return None
## 定义一个连接数据库并注册自定义函数的函数
def connect_db(db_path):
## 连接到指定路径的 SQLite 数据库
conn = sqlite3.connect(db_path)
## 将 Python 函数 'json_extract' 注册为自定义 SQL 函数
## "json_extract" 是 SQL 中的名称，2 是参数数量，json_extract 是 Python 函数
conn.create_function("json_extract", 2, json_extract)
## 返回数据库连接
return conn
## 当脚本直接执行时运行此块
if __name__ == '__main__':
## 连接到数据库并注册函数
conn = connect_db('mydatabase.db')
## 创建一个游标对象来执行 SQL 查询
cursor = conn.cursor()
## 使用自定义 SQL 函数从 'details' 列中提取 'brand'
cursor.execute("SELECT json_extract(details, 'brand') FROM products WHERE name = 'Laptop'")
## 获取结果并打印
print(cursor.fetchone()[0])
## 使用自定义 SQL 函数从嵌套的 'specs' 对象中提取 'cpu'
cursor.execute("SELECT json_extract(details, 'specs.cpu') FROM products WHERE name = 'Laptop'")
## 获取结果并打印
print(cursor.fetchone()[0])
## 关闭数据库连接
conn.close()
此 Python 代码定义了一个 json_extract 函数，该函数接受 JSON 字符串和路径作为输入，并返回该路径下的值。它还包含一个 connect_db 函数，用于连接到 SQLite 数据库并注册 json_extract 函数。
json.loads(json_str)：此行将 JSON 字符串解析为 Python 字典。path.split('.')：此行将路径拆分为组件列表。例如，'specs.cpu'变为['specs', 'cpu']。- 循环遍历路径组件，访问 JSON 数据中的嵌套值。
保存文件并退出 nano。
现在，运行 Python 脚本。
python3 json_extractor.py
预期输出：
Dell
Intel i7
此脚本连接到数据库，注册 json_extract 函数，然后使用它来提取 Laptop 的品牌和 CPU。
使用 JSON 查询过滤数据
在此步骤中，你将使用自定义的 json_extract 函数，根据 JSON 字段中的值来过滤数据。
再次打开 json_extractor.py 文件。
nano json_extractor.py
修改 json_extractor.py 文件，加入一个用于查询数据库的函数：
## 导入必要的库
import sqlite3
import json
## 定义一个函数，使用路径从 JSON 字符串中提取值
def json_extract(json_str, path):
try:
## 将 JSON 字符串解析为 Python 字典
json_data = json.loads(json_str)
## 将路径分割成组件（例如，'specs.cpu' 变成 ['specs', 'cpu']）
path_components = path.split('.')
## 从完整的 JSON 对象开始
value = json_data
## 使用路径组件遍历 JSON 对象
for component in path_components:
## 获取当前组件的值
value = value.get(component)
## 返回最终值
return value
## 处理 JSON 无效或路径不存在的错误
except (json.JSONDecodeError, AttributeError, TypeError):
return None
## 定义一个函数来连接数据库并注册自定义函数
def connect_db(db_path):
## 连接到指定路径的 SQLite 数据库
conn = sqlite3.connect(db_path)
## 将 Python 函数 'json_extract' 注册为自定义 SQL 函数
conn.create_function("json_extract", 2, json_extract)
## 返回数据库连接
return conn
## 定义一个函数来根据 JSON 字段过滤产品
def filter_products(db_path, json_path, value):
## 连接到数据库
conn = connect_db(db_path)
## 创建一个游标对象
cursor = conn.cursor()
## 使用 f-string 创建 SQL 查询，按 JSON 值进行过滤
query = f"SELECT * FROM products WHERE json_extract(details, '{json_path}') = '{value}'"
## 执行查询
cursor.execute(query)
## 获取所有匹配的结果
results = cursor.fetchall()
## 关闭数据库连接
conn.close()
## 返回结果
return results
## 当脚本直接执行时运行此块
if __name__ == '__main__':
## 示例用法：
## 过滤品牌为 'Dell' 的产品
dell_products = filter_products('mydatabase.db', 'brand', 'Dell')
print("Products with brand 'Dell':", dell_products)
## 过滤 CPU 为 'Intel i7' 的产品
intel_products = filter_products('mydatabase.db', 'specs.cpu', 'Intel i7')
print("Products with CPU 'Intel i7':", intel_products)
此代码添加了一个 filter_products 函数，该函数接受数据库路径、JSON 路径和值作为输入。然后它连接到数据库，注册 json_extract 函数，并执行一个查询来查找在指定 JSON 路径处的值与给定值匹配的所有产品。
保存文件并退出 nano。
现在，运行 Python 脚本。
python3 json_extractor.py
预期输出：
Products with brand 'Dell': [(1, 'Laptop', '{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}')]
Products with CPU 'Intel i7': [(1, 'Laptop', '{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}')]
此输出显示了符合指定条件的产品。
更新 JSON 值
在本步骤中，你将学习如何更新 JSON 字段中的值。
打开 json_extractor.py 文件。
nano json_extractor.py
修改 json_extractor.py 文件，加入用于更新 JSON 和数据库的函数：
## 导入必要的库
import sqlite3
import json
## 定义一个函数，使用路径从 JSON 字符串中提取值
def json_extract(json_str, path):
try:
## 将 JSON 字符串解析为 Python 字典
json_data = json.loads(json_str)
## 将路径分割成组件
path_components = path.split('.')
## 从完整的 JSON 对象开始
value = json_data
## 遍历 JSON 对象
for component in path_components:
value = value.get(component)
return value
except (json.JSONDecodeError, AttributeError, TypeError):
return None
## 定义一个函数来连接数据库并注册自定义函数
def connect_db(db_path):
## 连接到数据库
conn = sqlite3.connect(db_path)
## 注册自定义 SQL 函数
conn.create_function("json_extract", 2, json_extract)
return conn
## 定义一个函数，根据 JSON 字段过滤产品
def filter_products(db_path, json_path, value):
## 连接到数据库
conn = connect_db(db_path)
## 创建一个游标对象
cursor = conn.cursor()
## 创建用于按 JSON 值过滤的 SQL 查询
query = f"SELECT * FROM products WHERE json_extract(details, '{json_path}') = '{value}'"
## 执行查询
cursor.execute(query)
## 获取所有匹配的结果
results = cursor.fetchall()
## 关闭连接
conn.close()
return results
## 定义一个函数来更新 JSON 字符串中的值
def update_json_value(json_str, path, new_value):
try:
## 将 JSON 字符串解析为 Python 字典
json_data = json.loads(json_str)
## 将路径分割成组件
path_components = path.split('.')
## 导航到目标值的父级
target = json_data
for i in range(len(path_components) - 1):
target = target.get(path_components[i])
## 更新目标值
target[path_components[-1]] = new_value
## 将 Python 字典转换回 JSON 字符串
return json.dumps(json_data)
except (json.JSONDecodeError, AttributeError, TypeError):
## 如果发生错误，则返回原始 JSON 字符串
return json_str
## 定义一个函数来更新数据库中产品的详细信息
def update_product_details(db_path, product_name, json_path, new_value):
## 连接到数据库
conn = sqlite3.connect(db_path)
## 创建一个游标对象
cursor = conn.cursor()
## 获取指定产品的当前详细信息
## '?' 是一个占位符，用于防止 SQL 注入
cursor.execute("SELECT details FROM products WHERE name = ?", (product_name,))
result = cursor.fetchone()
## 如果产品不存在，则关闭连接并返回 False
if not result:
conn.close()
return False
## 获取当前的 JSON 详细信息字符串
current_details = result[0]
## 使用新值更新 JSON 字符串
updated_details = update_json_value(current_details, json_path, new_value)
## 使用新 JSON 字符串更新数据库
cursor.execute("UPDATE products SET details = ? WHERE name = ?", (updated_details, product_name))
## 提交数据库更改
conn.commit()
## 关闭连接
conn.close()
return True
## 当脚本直接执行时，此块将运行
if __name__ == '__main__':
## 示例用法：将笔记本电脑的内存更新为 32GB
update_product_details('mydatabase.db', 'Laptop', 'specs.memory', '32GB')
print("Laptop memory updated to 32GB")
## 通过再次获取数据来验证更新
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()
cursor.execute("SELECT details FROM products WHERE name = 'Laptop'")
## 获取更新后的详细信息
updated_details = cursor.fetchone()[0]
print("Updated Laptop details:", updated_details)
## 关闭连接
conn.close()
此代码添加了两个函数：
update_json_value: 此函数接受一个 JSON 字符串、一个路径和一个新值作为输入。它解析 JSON 字符串，在指定路径更新值，并返回更新后的 JSON 字符串。update_product_details: 此函数接受一个数据库路径、一个产品名称、一个 JSON 路径和一个新值作为输入。它连接到数据库，检索产品的当前 JSON 数据，使用update_json_value在指定路径更新值，然后使用修改后的 JSON 数据更新数据库。
保存文件并退出 nano。
现在，运行 Python 脚本。
python3 json_extractor.py
预期输出：
Laptop memory updated to 32GB
Updated Laptop details: {"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "32GB", "storage": "512GB SSD"}}
此输出确认笔记本电脑的内存已在数据库中更新为 32GB。
总结
在这个实验中，你已经学习了如何在 SQLite 中处理 JSON 数据。你首先创建了一个数据库和一个表来存储 JSON 数据。然后，你学习了如何将 JSON 数据插入到表中。你创建了一个自定义的 Python 函数来从 JSON 数据中提取特定字段，并使用此函数根据 JSON 字段中的值过滤数据。最后，你学习了如何使用自定义的 Python 函数更新 JSON 字段中的值。这些技能为在 SQLite 数据库中有效地管理 JSON 数据奠定了基础。
