求Bonobo从MySQL(SQLAlchemy)读取后生成XML的自定义写入器示例

嘿，恭喜你已经搞定了Bonobo和SQLAlchemy的数据流读取部分！写自定义XML写入器其实没那么复杂，我给你一个新手友好的极简示例，一步步帮你实现：

核心思路 #

Bonobo的转换步骤本质是可迭代的函数/类，我们可以写一个类来处理XML文件的初始化、逐行写入和收尾工作——逻辑很清晰：先写XML头部和根标签，每收到一行数据就生成对应的XML元素，最后闭合根标签。

示例代码 #

1. 自定义XML写入器类（基础版）

这个版本用字符串拼接实现，不用额外学复杂的XML库，对新手最友好：

import bonobo
from bonobo_sqlalchemy import Select
from sqlalchemy import create_engine
from html import escape  # 用来转义XML特殊字符（比如&、<、>）

class SimpleXMLWriter:
    def __init__(self, filename):
        # 初始化XML文件，写入头部和根标签
        self.file = open(filename, 'w', encoding='utf-8')
        self.file.write('<?xml version="1.0" encoding="UTF-8"?>\n')
        self.file.write('<user_data>\n')
        
        # 这里替换成你的UserData表的实际字段名（和SELECT *的顺序一致）
        self.field_names = ['id', 'username', 'email', 'create_time']

    def __call__(self, *row):
        # 将Bonobo传递的行数据（元组）转成字典
        user_dict = dict(zip(self.field_names, row))
        
        # 拼接单个用户的XML元素
        user_xml = '  <user>\n'
        for key, value in user_dict.items():
            # 转义特殊字符，避免XML格式错误
            safe_value = escape(str(value))
            user_xml += f'    <{key}>{safe_value}</{key}>\n'
        user_xml += '  </user>\n'
        
        # 写入文件
        self.file.write(user_xml)

    def close(self):
        # 流程结束后闭合根标签并关闭文件
        self.file.write('</user_data>\n')
        self.file.close()

2. 整合到Bonobo流程中

把自定义写入器和你的数据库读取流程串起来：

def get_graph():
    graph = bonobo.Graph()
    
    # 替换成你的MySQL实际连接信息
    db_engine = create_engine('mysql+pymysql://your_username:your_password@localhost:3306/your_db_name')
    
    # 初始化XML写入器
    xml_writer = SimpleXMLWriter('user_output.xml')
    
    # 搭建数据流链：读数据库 → 写XML
    graph.add_chain(
        Select('SELECT * FROM UserData', engine=db_engine),
        xml_writer,
    )
    
    return graph, xml_writer

if __name__ == '__main__':
    # 运行Bonobo流程
    parser = bonobo.get_argument_parser()
    with bonobo.parse_args(parser) as options:
        graph, xml_writer = get_graph()
        bonobo.run(graph, services=options.services)
    
    # 最后记得关闭文件
    xml_writer.close()

进阶优化（可选） #

如果想更安全地管理文件资源，可以用上下文管理器改造写入器，不用手动调用close()：

class SafeXMLWriter:
    def __init__(self, filename):
        self.filename = filename
        self.file = None
        self.field_names = ['id', 'username', 'email', 'create_time']

    def __enter__(self):
        self.file = open(self.filename, 'w', encoding='utf-8')
        self.file.write('<?xml version="1.0" encoding="UTF-8"?>\n')
        self.file.write('<user_data>\n')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 无论流程是否报错，都会自动闭合根标签并关闭文件
        if self.file:
            self.file.write('</user_data>\n')
            self.file.close()

    def __call__(self, *row):
        # 和基础版的处理逻辑一致
        user_dict = dict(zip(self.field_names, row))
        user_xml = '  <user>\n'
        for key, value in user_dict.items():
            safe_value = escape(str(value))
            user_xml += f'    <{key}>{safe_value}</{key}>\n'
        user_xml += '  </user>\n'
        self.file.write(user_xml)

使用时只需要用with语句包裹：

def get_graph():
    graph = bonobo.Graph()
    db_engine = create_engine('mysql+pymysql://your_username:your_password@localhost:3306/your_db_name')
    
    with SafeXMLWriter('user_output.xml') as writer:
        graph.add_chain(
            Select('SELECT * FROM UserData', engine=db_engine),
            writer,
        )
    
    return graph

新手注意事项 #

• 一定要替换field_names为你的UserData表的实际字段名，顺序要和SELECT *返回的一致
• 数据里如果有XML特殊字符（比如&、<、>），必须用html.escape转义，否则生成的XML会无效
• 数据库连接字符串里的your_username、your_password等信息要换成自己的实际配置

内容的提问来源于stack exchange，提问作者user3500192