XML是一种常用的结构化数据交换格式,SQLite是轻量级的嵌入式数据库,二者结合可以满足很多本地数据存储的需求。Python标准库中提供了xml.etree.ElementTree模块用于解析XML,sqlite3模块用于操作SQLite数据库,无需安装额外依赖就能完成数据解析和存储的全流程。
准备工作
首先需要准备一个待解析的XML文件,假设我们有一个存储用户信息的user_data.xml文件,内容如下:
<?xml version="1.0" encoding="UTF-8"?>
<users>
<user>
<id>1</id>
<name>张三</name>
<age>25</age>
<email>zhangsan@ipipp.com</email>
</user>
<user>
<id>2</id>
<name>李四</name>
<age>28</age>
<email>lisi@ipipp.com</email>
</user>
</users>
解析XML数据
使用xml.etree.ElementTree模块读取并解析XML文件,提取需要的节点数据。解析的核心步骤是先加载XML文件,再通过迭代的方式获取每个user节点的子元素内容。
import xml.etree.ElementTree as ET
# 加载XML文件
tree = ET.parse('user_data.xml')
# 获取根节点
root = tree.getroot()
# 存储解析后的用户数据
user_list = []
# 遍历所有user节点
for user in root.findall('user'):
user_id = user.find('id').text
name = user.find('name').text
age = user.find('age').text
email = user.find('email').text
# 将数据存入列表
user_list.append({
'id': user_id,
'name': name,
'age': age,
'email': email
})
# 打印解析结果验证
for item in user_list:
print(item)
创建SQLite数据库表
使用sqlite3模块连接SQLite数据库,如果数据库文件不存在会自动创建,然后创建对应的用户表结构,表字段需要和XML解析出的数据字段对应。
import sqlite3
# 连接SQLite数据库,文件不存在会自动创建
conn = sqlite3.connect('user.db')
# 创建游标对象
cursor = conn.cursor()
# 创建用户表,字段类型和XML中的数据对应
create_table_sql = '''
CREATE TABLE IF NOT EXISTS user (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
email TEXT
)
'''
cursor.execute(create_table_sql)
# 提交事务
conn.commit()
将解析数据存入SQLite
遍历之前解析得到的用户数据列表,将每条数据插入到SQLite的用户表中,使用参数化查询可以避免SQL注入问题,同时提高插入效率。
# 遍历用户数据列表,插入到数据库
insert_sql = 'INSERT INTO user (id, name, age, email) VALUES (?, ?, ?, ?)'
for user in user_list:
# 按字段顺序传入参数
cursor.execute(insert_sql, (user['id'], user['name'], user['age'], user['email']))
# 提交事务,保存插入的数据
conn.commit()
# 关闭游标和连接
cursor.close()
conn.close()
常见问题说明
编码问题
如果XML文件包含中文,需要确保XML声明中的encoding属性和文件实际编码一致,否则解析时会出现乱码。如果解析时出现编码错误,可以在打开文件时指定编码格式:
# 指定编码打开XML文件
with open('user_data.xml', 'r', encoding='utf-8') as f:
tree = ET.parse(f)
数据校验
插入数据前可以对解析出的数据进行校验,比如判断id是否为数字、邮箱格式是否合法,避免无效数据存入数据库。如果需要批量插入大量数据,可以使用executemany方法提高效率:
# 批量插入数据 cursor.executemany(insert_sql, [(u['id'], u['name'], u['age'], u['email']) for u in user_list])
验证存储结果
可以重新连接数据库查询数据,验证是否存储成功:
# 重新连接数据库查询数据
conn = sqlite3.connect('user.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM user')
rows = cursor.fetchall()
for row in rows:
print(row)
cursor.close()
conn.close()