如何通过Azure Data Factory将CSV数据导入API？求参考示例

以下是用Python实现从CSV读取客户数据、转换格式后调用API创建客户的完整实用示例：

前提准备 #

先安装必要依赖：

pip install requests

假设你的客户CSV文件customers.csv结构如下：

name,email,phone
Alice Smith,alice@example.com,1234567890
Bob Johnson,bob@example.com,0987654321

实现代码 #

import csv
import requests

# 替换为你的API实际配置
API_ENDPOINT = "https://your-api-domain.com/api/customers"
AUTH_TOKEN = "your-api-auth-token"
REQUEST_HEADERS = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {AUTH_TOKEN}"
}

def import_csv_to_api(csv_file_path):
    with open(csv_file_path, mode='r', encoding='utf-8') as csv_file:
        # 直接将CSV行转为字典，自动匹配表头
        csv_reader = csv.DictReader(csv_file)
        
        for customer_row in csv_reader:
            # 根据API要求映射字段
            api_payload = {
                "full_name": customer_row["name"],
                "email": customer_row["email"],
                "phone_number": customer_row["phone"]
            }
            
            try:
                # 发送POST请求创建客户
                response = requests.post(API_ENDPOINT, json=api_payload, headers=REQUEST_HEADERS)
                response.raise_for_status()  # 触发HTTP错误状态码异常
                print(f"✅ 客户创建成功: {customer_row['name']} (ID: {response.json().get('customer_id')})")
            except requests.exceptions.RequestException as err:
                print(f"❌ 客户创建失败 {customer_row['name']}: {str(err)}")

if __name__ == "__main__":
    import_csv_to_api("customers.csv")

关键细节说明 #

• CSV解析：csv.DictReader自动将表头作为字典键，省去手动索引列的麻烦。
• 字段映射：务必根据目标API的请求体结构，调整api_payload中的字段对应关系。
• 认证适配：示例用Bearer Token认证，若API用API Key，直接替换Authorization为"X-API-Key": "your-key"即可。
• 错误容错：捕获请求异常，保证单个客户创建失败不会中断整个导入流程。

扩展优化方向 #

• 批量提交：如果API支持批量创建，可积累N条数据后一次性发送，减少请求次数。
• 数据校验：添加邮箱格式、手机号规则等校验逻辑，避免无效数据发送到API。
• 日志记录：用logging模块替代print，生成结构化日志方便后续排查问题。

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