Azure Data Factory Python SDK如何将PipelineResource转换为JSON文件

失败原因 #

直接调用json.loads(pipeline_object)、json.dumps(pipeline_object)失败的核心原因是PipelineResource属于Azure SDK自定义模型类型，不属于Python原生支持JSON序列化的基础类型，JSON模块默认无法识别其内部结构。

可用实现方法 #

方法1：使用模型内置方法转换（兼容性最好） #

所有Azure SDK模型类都内置as_dict()方法，可以先将模型对象转换为结构完全对齐Azure官方API规范的Python字典，再通过json模块写入文件，仅需额外处理datetime等特殊类型的序列化逻辑即可：

import json
from datetime import datetime

# 自定义JSON序列化处理函数，解决日期类型无法序列化的问题
def serialize_datetime(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Unserializable type: {type(obj)}")

# pipeline_object为你已创建的PipelineResource实例
pipeline_dict = pipeline_object.as_dict()

# 写入JSON文件
with open("adf_pipeline.json", "w", encoding="utf-8") as f:
    json.dump(
        pipeline_dict,
        f,
        indent=2,
        default=serialize_datetime,
        ensure_ascii=False
    )

该方法导出的JSON结构和Azure门户导出的ADF管道结构完全一致，可直接用于ARM模板编写、管道批量导入等场景。

方法2：使用SDK内置JSON编码器（写法更简洁） #

如果使用的是2.0.0及以上版本的azure-mgmt-datafactory包，可以直接调用azure.core提供的官方JSON编码器，无需手动编写特殊类型处理逻辑，直接对模型对象做序列化：

import json
from azure.core.serialization import AzureJSONEncoder

# pipeline_object为你已创建的PipelineResource实例
with open("adf_pipeline.json", "w", encoding="utf-8") as f:
    json.dump(
        pipeline_object,
        f,
        indent=2,
        cls=AzureJSONEncoder,
        ensure_ascii=False
    )

注意事项 #

• 不要直接对模型实例调用json.dumps()且不指定自定义编码器，必然会触发类型不支持序列化的报错。
• 导出的JSON文件如果需要反向加载为PipelineResource对象，可以直接调用模型类的from_dict()方法传入读取到的字典即可，无需手动做字段映射。

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