大家好，今天为大家分享一个有趣的 Python 库 - xmltodict。

Github地址：https://github.com/martinblech/xmltodict

xmltodict是一个轻量级且功能强大的Python第三方库，其设计理念是让XML数据处理变得如同操作JSON一样直观简单。该库基于高性能的Expat解析器构建，能够快速将XML文档转换为Python字典结构，也支持将字典数据逆向转换为XML格式。xmltodict最大的优势在于其简洁的API设计和出色的性能表现，特别适合处理复杂的XML数据转换任务。

安装

1、安装方法

xmltodict可以通过多种方式进行安装，最常用的方法是使用pip包管理器：

pip install xmltodict

对于Python 3环境，推荐使用：

pip3 install xmltodict

在系统级别安装可使用各操作系统的包管理器：

# Ubuntu/Debian系统
sudo apt-get install python3-xmltodict

# Fedora系统
sudo dnf install python-xmltodict

# openSUSE系统
sudo zypper in python3-xmltodict

# 使用conda安装
conda install anaconda::xmltodict

2、验证安装

安装完成后，可以通过以下命令验证xmltodict是否正确安装：

import xmltodict
print("xmltodict安装成功!")

# 快速测试基本功能
test_xml = "<root><item>test</item></root>"
result = xmltodict.parse(test_xml)
print(result)  # 应输出: {'root': {'item': 'test'}}

特性

• JSON风格的XML处理：将XML数据转换为类似JSON的Python字典结构，操作直观简便
• 双向转换支持：提供parse()和unparse()方法，支持XML到字典以及字典到XML的双向转换
• 高性能解析器：基于Expat解析器构建，处理速度快，内存占用低
• 流式处理模式：支持大型XML文件的流式解析，适合处理GB级别的XML数据
• 属性和文本处理：智能处理XML属性和文本内容，使用@前缀标识属性，#text标识文本内容
• 命名空间支持：提供完整的XML命名空间处理能力，支持命名空间展开和折叠

基本功能

1、XML到字典的转换

下面的代码示例展示了xmltodict最核心的功能：将XML文档解析为Python字典。这个功能特别适用于需要处理API响应数据、配置文件或任何结构化XML内容的场景。

import xmltodict
import json

# 复杂XML示例
xml_data = """
<bookstore>
    <book id="1" category="fiction">
        <title lang="en">Great Gatsby</title>
        <author>F. Scott Fitzgerald</author>
        <price currency="USD">10.99</price>
    </book>
    <book id="2" category="science">
        <title lang="en">Brief History of Time</title>
        <author>Stephen Hawking</author>
        <price currency="USD">15.99</price>
    </book>
</bookstore>
"""

# 解析XML为字典
parsed_data = xmltodict.parse(xml_data)

# 美化输出查看结构
print(json.dumps(parsed_data, indent=2, ensure_ascii=False))

# 访问特定数据
books = parsed_data['bookstore']['book']
for book in books:
    print(f"书名: {book['title']['#text']}")
    print(f"作者: {book['author']}")
    print(f"价格: {book['price']['#text']} {book['price']['@currency']}")

运行结果：

{
  "bookstore": {
    "book": [
      {
        "@id": "1",
        "@category": "fiction",
        "title": {
          "@lang": "en",
          "#text": "Great Gatsby"
        },
        "author": "F. Scott Fitzgerald",
        "price": {
          "@currency": "USD",
          "#text": "10.99"
        }
      },
      {
        "@id": "2",
        "@category": "science",
        "title": {
          "@lang": "en",
          "#text": "Brief History of Time"
        },
        "author": "Stephen Hawking",
        "price": {
          "@currency": "USD",
          "#text": "15.99"
        }
      }
    ]
  }
}
书名: Great Gatsby
作者: F. Scott Fitzgerald
价格: 10.99 USD
书名: Brief History of Time
作者: Stephen Hawking
价格: 15.99 USD

2、字典到XML的转换

以下代码演示了如何将Python字典数据转换回XML格式。xmltodict的unparse方法支持美化输出，能够生成格式整齐、易于阅读的XML文档。

import xmltodict

# 构建字典数据
user_data = {
    'user': {
        '@id': '12345',
        'profile': {
            'name': '张三',
            'email': 'zhangsan@example.com',
            'preferences': {
                'language': 'zh-CN',
                'theme': 'dark',
                'notifications': {
                    '@enabled': 'true',
                    '#text': 'email'
                }
            }
        },
        'settings': {
            'privacy': 'public',
            'location': {
                '@visible': 'false',
                '#text': 'Beijing'
            }
        }
    }
}

# 转换为XML格式
xml_output = xmltodict.unparse(user_data, pretty=True)
print("生成的XML:")
print(xml_output)

# 保存到文件
with open('user_config.xml', 'w', encoding='utf-8') as f:
    f.write(xml_output)

运行结果：

生成的XML:
<?xml version="1.0" encoding="utf-8"?>
<user id="12345">
	<profile>
		<name>张三</name>
		<email>zhangsan@example.com</email>
		<preferences>
			<language>zh-CN</language>
			<theme>dark</theme>
			<notifications enabled="true">email</notifications>
		</preferences>
	</profile>
	<settings>
		<privacy>public</privacy>
		<location visible="false">Beijing</location>
	</settings>
</user>

高级功能

1、命名空间处理

xmltodict提供了强大的XML命名空间处理能力，允许开发者灵活控制命名空间的展开和折叠：

import json

import xmltodict

# 包含命名空间的XML
namespaced_xml = """
<root xmlns="http://defaultns.com/"
      xmlns:books="http://books.com/"
      xmlns:authors="http://authors.com/">
    <title>Library Catalog</title>
    <books:book>
        <books:title>Python Programming</books:title>
        <authors:author>John Doe</authors:author>
    </books:book>
</root>
"""

# 启用命名空间处理
parsed_with_ns = xmltodict.parse(
    namespaced_xml,
    process_namespaces=True
)

print("启用命名空间处理的结果:")
print(json.dumps(parsed_with_ns, indent=2))

# 自定义命名空间映射
namespace_map = {
    'http://defaultns.com/': None,  # 跳过默认命名空间
    'http://books.com/': 'bk',      # 简化books命名空间
    'http://authors.com/': 'auth'   # 简化authors命名空间
}

parsed_custom_ns = xmltodict.parse(
    namespaced_xml,
    process_namespaces=True,
    namespaces=namespace_map
)

print("\n自定义命名空间映射结果:")
print(json.dumps(parsed_custom_ns, indent=2))

运行结果：

启用命名空间处理的结果:
{
  "http://defaultns.com/:root": {
    "http://defaultns.com/:title": "Library Catalog",
    "http://books.com/:book": {
      "http://books.com/:title": "Python Programming",
      "http://authors.com/:author": "John Doe"
    }
  }
}

自定义命名空间映射结果:
{
  "root": {
    "title": "Library Catalog",
    "bk:book": {
      "bk:title": "Python Programming",
      "auth:author": "John Doe"
    }
  }
}

2、流式处理大文件

对于大型XML文件，xmltodict提供了流式处理模式，能够显著降低内存使用量：

import xml.etree.ElementTree as ET
from io import BytesIO


def process_large_xml_stream():
    """演示流式处理大型XML文件"""

    # 模拟大型XML数据
    large_xml = """
    <catalog>
        <products>
            <product id="1">
                <name>Laptop</name>
                <price>999.99</price>
            </product>
            <product id="2">
                <name>Mouse</name>
                <price>29.99</price>
            </product>
            <product id="3">
                <name>Keyboard</name>
                <price>79.99</price>
            </product>
        </products>
    </catalog>
    """

    # 将字符串转换为字节流
    xml_bytes = large_xml.encode('utf-8')
    xml_stream = BytesIO(xml_bytes)

    # 使用iterparse进行流式解析
    context = ET.iterparse(xml_stream, events=('start', 'end'))

    for event, elem in context:
        if event == 'end' and elem.tag == 'product':
            # 处理产品数据
            product_id = elem.get('id')
            name = elem.find('name').text
            price = elem.find('price').text

            print(f"处理产品 ID: {product_id}")
            print(f"产品名称: {name}")
            print(f"产品价格: {price}")
            print("-" * 30)

            # 清除已处理的元素以节省内存
            elem.clear()

    # 显式关闭流（虽然不是必须的，但这是好习惯）
    xml_stream.close()


if __name__ == "__main__":
    process_large_xml_stream()

运行结果：

处理产品 ID: 1
产品名称: Laptop
产品价格: 999.99
------------------------------
处理产品 ID: 2
产品名称: Mouse
产品价格: 29.99
------------------------------
处理产品 ID: 3
产品名称: Keyboard
产品价格: 79.99
------------------------------

实际应用场景

1、Web服务数据转换

在现代Web开发中，经常需要处理来自不同系统的XML数据并转换为JSON格式以便前端使用。xmltodict能够简化这个转换过程，特别适用于企业级系统集成和API数据格式转换场景。

import xmltodict
import json
from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)


class XMLDataProcessor:
    """XML数据处理服务"""

    def __init__(self):
        self.supported_formats = ['json', 'dict']

    def process_soap_response(self, soap_xml):
        """处理SOAP响应数据"""
        try:
            # 添加默认命名空间处理
            parsed_data = xmltodict.parse(
                soap_xml,
                process_namespaces=True,
                namespaces={
                    'http://schemas.xmlsoap.org/soap/envelope/': 'soap',
                    None: 'ns'  # 处理无命名空间的元素
                }
            )

            # 更健壮的SOAP结构提取
            envelope = parsed_data.get('soap:Envelope', parsed_data.get('Envelope', {}))
            body = envelope.get('soap:Body', envelope.get('Body', {}))

            # 移除SOAP包装，只保留业务数据
            business_data = next(
                (value for key, value in body.items()
                 if not key.startswith(('soap:', '@'))),
                {}
            )

            return {
                'success': True,
                'data': business_data,
                'metadata': {
                    'soap_version': '1.1',
                    'processed_at': datetime.datetime.now().isoformat()
                }
            }

        except Exception as e:
            return {
                'success': False,
                'error': f"SOAP处理失败: {str(e)}",
                'data': None
            }

    def convert_payment_notification(self, payment_xml):
        """转换支付通知XML"""
        try:
            payment_dict = xmltodict.parse(
                payment_xml,
                force_cdata=True,  # 处理可能包含特殊字符的字段
                attr_prefix=''
            )

            # 更安全的字段提取
            payment = payment_dict.get('payment', {})

            # 标准化支付数据格式
            standardized = {
                'transaction': {
                    'id': payment.get('id'),
                    'amount': self._safe_float(payment.get('amount')),
                    'currency': payment.get('currency', 'CNY'),
                    'status': payment.get('status'),
                    'timestamp': payment.get('timestamp'),
                },
                'merchant': payment.get('merchant', {}),
                'raw_data': payment  # 保留原始数据
            }

            return {
                'success': True,
                'data': standardized
            }

        except Exception as e:
            return {
                'success': False,
                'error': f"支付通知处理失败: {str(e)}",
                'data': None
            }

    def _safe_float(self, value):
        """安全转换为float"""
        try:
            return float(value) if value else 0.0
        except (ValueError, TypeError):
            return 0.0


def validate_xml_content(f):
    """XML内容验证装饰器"""

    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not request.data:
            return jsonify({
                'success': False,
                'error': '未提供XML内容',
                'data': None
            }), 400

        try:
            request.get_data().decode('utf-8')
        except UnicodeDecodeError:
            return jsonify({
                'success': False,
                'error': '无效的XML编码(仅支持UTF-8)',
                'data': None
            }), 400

        return f(*args, **kwargs)

    return decorated_function


@app.route('/convert/soap', methods=['POST'])
@validate_xml_content
def convert_soap():
    """SOAP XML转换API端点"""
    xml_content = request.data.decode('utf-8')
    processor = XMLDataProcessor()
    result = processor.process_soap_response(xml_content)
    return jsonify(result)


@app.route('/convert/payment', methods=['POST'])
@validate_xml_content
def convert_payment():
    """支付通知XML转换端点"""
    xml_content = request.data.decode('utf-8')
    processor = XMLDataProcessor()
    result = processor.convert_payment_notification(xml_content)
    return jsonify(result)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

2、配置文件管理系统

在企业应用中，经常需要管理复杂的XML配置文件。xmltodict可以轻松实现配置文件的读取、修改和保存操作。

import xmltodict
import json
import os
import threading
from datetime import datetime
from typing import Dict, Any, Optional, Union


class ConfigurationManager:
    """XML配置文件管理器"""

    def __init__(self, config_path: str):
        self.config_path = config_path
        self.config_data: Dict[str, Any] = {}
        self.backup_dir = 'config_backups'
        self.lock = threading.Lock()
        self._ensure_backup_dir()

    def _ensure_backup_dir(self) -> None:
        """确保备份目录存在"""
        with self.lock:
            if not os.path.exists(self.backup_dir):
                os.makedirs(self.backup_dir, exist_ok=True)

    def _validate_config_structure(self) -> bool:
        """验证配置数据结构有效性"""
        return isinstance(self.config_data, dict)

    def load_configuration(self) -> Dict[str, Any]:
        """加载XML配置文件"""
        try:
            with self.lock, open(self.config_path, 'r', encoding='utf-8') as file:
                xml_content = file.read()
                if not xml_content.strip():
                    raise ValueError("配置文件为空")

                self.config_data = xmltodict.parse(xml_content)

                if not self._validate_config_structure():
                    raise ValueError("无效的配置文件结构")

                return self.config_data

        except FileNotFoundError:
            print(f"[ERROR] 配置文件未找到: {self.config_path}")
            return {}
        except Exception as e:
            print(f"[ERROR] 配置文件加载失败: {str(e)}")
            return {}

    def backup_configuration(self) -> str:
        """备份当前配置"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_filename = f"config_backup_{timestamp}.xml"
        backup_path = os.path.join(self.backup_dir, backup_filename)

        try:
            with self.lock:
                if not os.path.exists(self.config_path):
                    raise FileNotFoundError("原始配置文件不存在")

                with open(self.config_path, 'r', encoding='utf-8') as source, \
                        open(backup_path, 'w', encoding='utf-8') as backup:
                    backup.write(source.read())

                print(f"[INFO] 配置已备份到: {backup_path}")
                return backup_path

        except Exception as e:
            print(f"[ERROR] 备份失败: {str(e)}")
            return ""

    def update_database_config(self, host: str, port: int,
                               username: str, password: str,
                               database: str) -> bool:
        """更新数据库配置"""
        try:
            with self.lock:
                if 'application' not in self.config_data:
                    self.config_data['application'] = {}

                self.config_data['application']['database'] = {
                    '@type': 'mysql',
                    'connection': {
                        'host': host,
                        'port': str(port),
                        'username': username,
                        'password': password,
                        'database': database,
                        'pool': {
                            '@size': '10',
                            '@timeout': '30'
                        }
                    }
                }
                return True
        except Exception as e:
            print(f"[ERROR] 更新数据库配置失败: {str(e)}")
            return False

    def update_logging_config(self, level: str, log_file: str,
                              max_size: str = '100MB') -> bool:
        """更新日志配置"""
        try:
            with self.lock:
                if 'application' not in self.config_data:
                    self.config_data['application'] = {}

                self.config_data['application']['logging'] = {
                    '@enabled': 'true',
                    'level': level,
                    'file': {
                        '@path': log_file,
                        '@maxSize': max_size,
                        '@backup': 'true'
                    },
                    'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
                }
                return True
        except Exception as e:
            print(f"[ERROR] 更新日志配置失败: {str(e)}")
            return False

    def save_configuration(self) -> bool:
        """保存配置到文件"""
        try:
            with self.lock:
                # 先备份当前配置
                self.backup_configuration()

                # 验证配置数据
                if not self._validate_config_structure():
                    raise ValueError("无效的配置数据结构")

                # 生成新的XML内容
                xml_content = xmltodict.unparse(
                    self.config_data,
                    pretty=True,
                    encoding='utf-8'
                )

                # 创建临时文件
                temp_path = f"{self.config_path}.tmp"
                with open(temp_path, 'w', encoding='utf-8') as file:
                    file.write(xml_content)

                # 原子性替换原文件
                os.replace(temp_path, self.config_path)

                print(f"[INFO] 配置已保存到: {self.config_path}")
                return True

        except Exception as e:
            print(f"[ERROR] 配置保存失败: {str(e)}")
            if os.path.exists(temp_path):
                os.remove(temp_path)
            return False

    def get_config_summary(self) -> Dict[str, Any]:
        """获取配置摘要"""
        summary = {
            'database_configured': False,
            'logging_configured': False,
            'total_sections': 0,
            'last_modified': None,
            'backup_count': 0,
            'config_size': 0
        }

        try:
            with self.lock:
                # 文件信息
                if os.path.exists(self.config_path):
                    stat = os.stat(self.config_path)
                    summary['last_modified'] = datetime.fromtimestamp(
                        stat.st_mtime
                    ).isoformat()
                    summary['config_size'] = stat.st_size

                # 备份文件计数
                if os.path.exists(self.backup_dir):
                    summary['backup_count'] = len([
                        f for f in os.listdir(self.backup_dir)
                        if f.endswith('.xml')
                    ])

                # 配置内容
                if self.config_data and 'application' in self.config_data:
                    app_config = self.config_data['application']
                    summary.update({
                        'database_configured': 'database' in app_config,
                        'logging_configured': 'logging' in app_config,
                        'total_sections': len(app_config)
                    })

                return summary
        except Exception as e:
            print(f"[ERROR] 获取配置摘要失败: {str(e)}")
            return summary

    def cleanup_backups(self, max_backups: int = 5) -> bool:
        """清理旧的备份文件"""
        try:
            with self.lock:
                if not os.path.exists(self.backup_dir):
                    return True

                backups = sorted([
                    os.path.join(self.backup_dir, f)
                    for f in os.listdir(self.backup_dir)
                    if f.endswith('.xml')
                ], key=os.path.getmtime, reverse=True)

                for old_backup in backups[max_backups:]:
                    os.remove(old_backup)
                    print(f"[INFO] 已删除旧备份: {old_backup}")

                return True
        except Exception as e:
            print(f"[ERROR] 清理备份失败: {str(e)}")
            return False


# 使用示例
if __name__ == '__main__':
    # 创建配置管理器
    config_manager = ConfigurationManager('app_config.xml')

    # 加载现有配置
    config_data = config_manager.load_configuration()
    print("当前配置:", json.dumps(config_data, indent=2, ensure_ascii=False))

    # 更新数据库配置
    success = config_manager.update_database_config(
        host='localhost',
        port=3306,
        username='app_user',
        password='secure_password',
        database='production_db'
    )
    print(f"数据库配置更新{'成功' if success else '失败'}")

    # 更新日志配置
    success = config_manager.update_logging_config(
        level='DEBUG',
        log_file='/var/log/application.log',
        max_size='500MB'
    )
    print(f"日志配置更新{'成功' if success else '失败'}")

    # 保存配置
    success = config_manager.save_configuration()
    print(f"配置保存{'成功' if success else '失败'}")

    # 显示配置摘要
    summary = config_manager.get_config_summary()
    print("\n配置摘要:")
    print(json.dumps(summary, indent=2, ensure_ascii=False))

    # 清理备份
    config_manager.cleanup_backups(max_backups=3)https://archive.biliimg.com/bfs/archive/87ab28155935de0788f38b1f2d69e26024bc39db.jpg

总结

Python xmltodict库作为一个专业的XML处理工具，成功地将复杂的XML操作简化为直观的字典操作，极大地提升了Python开发者处理XML数据的效率和体验。该库基于高性能的Expat解析器构建，不仅保证了处理速度，还通过流式处理模式有效解决了大文件处理的内存问题。其双向转换能力、完整的命名空间支持和灵活的配置选项，使其在Web服务集成、配置文件管理、数据格式转换等多种应用场景中都能发挥重要作用。