如何在Azure Data Factory中实现带Token验证的API数据请求

需求概述 #

需要连接指定API获取数据，API要求通过accountCode和密钥生成有效Token才能访问。Token生成逻辑如下：

1. 生成当前毫秒级时间戳，加上1年的TTL（31536000000毫秒）得到dateToken
2. 构造包含dateToken的预请求URL
3. 将预请求URL与密钥拼接后做MD5哈希，得到最终的Token
4. 拼接完整请求URL，包含dateToken和token参数

API的BaseURL格式为：http://api.xxxxx.com/{accountCode}/data（需替换{accountCode}为实际值）

实现方法 #

方法一：使用Python自定义活动（直接复用现有代码逻辑） #

这种方法将现成的Python代码封装为ADF自定义活动，运行在Integration Runtime上，逻辑直接清晰。

步骤1：修改Python脚本适配ADF

将原有Python代码调整为可接收ADF传入参数的版本，示例代码如下：

import time
import requests
from hashlib import md5
import sys

# 从ADF管道参数读取accountCode、API密钥、输出路径
account_code = sys.argv[1]
api_key = sys.argv[2]
output_file_path = sys.argv[3]

# 生成毫秒级时间戳与dateToken
actual_unix_time = int(time.time_ns() / 1000)
TTL = 31536000000  # 1年有效期
expiration_time = actual_unix_time + TTL

# 构造请求路径与预URL
base_url = "https://api.xxxxx.com"
api_path = f"/{account_code}/data?fromDate=last6Hours&granularity=minute&type=ALL%2CVOD%2CLIVE&operation=reduceJoin&metrics=bufferratio"
pre_url = f"{api_path}&dateToken={expiration_time}"

# 生成MD5 Token
token_generated = md5(f"{pre_url}{api_key}".encode('utf-8'))
token_value = token_generated.hexdigest()

# 拼接完整请求URL并发送请求
request_url = f"{base_url}{pre_url}&token={token_value}"
response = requests.get(request_url)
response.raise_for_status()  # 捕获请求错误

# 将响应结果写入指定路径（如ADLS Gen2或Blob存储）
with open(output_file_path, 'w', encoding='utf-8') as f:
    f.write(response.text)

print("API数据获取成功")

步骤2：配置Integration Runtime

• 使用Self-Hosted Integration Runtime（需确保本地环境安装Python 3.6+及requests库），或Azure Integration Runtime（需提前安装依赖库，或打包依赖为wheel文件上传）。

步骤3：创建并配置自定义活动

1. 在ADF管道中添加自定义活动，关联已配置的Integration Runtime。
2. 在设置标签页：
   • 选择脚本存储位置（如Azure Blob存储），上传修改后的Python脚本。
   • 设置命令为python，参数为@concat(pipeline().parameters.accountCode, ' ', pipeline().parameters.apiKey, ' ', pipeline().parameters.outputFilePath)（参数从管道参数传入，建议密钥从Azure Key Vault获取）。
3. 运行管道，脚本会自动生成Token、调用API并保存结果。

方法二：使用Web活动+Azure Functions生成Token #

如果不想维护自定义脚本，可将Token生成逻辑封装到Azure Functions，再通过ADF Web活动调用获取Token，最终调用目标API。

步骤1：创建Azure Function（生成Token）

创建HTTP触发的Function，输入accountCode、apiKey等参数，返回完整请求URL。示例C#代码：

using System;
using System.Net;
using System.Security.Cryptography;
using System.Text;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Extensions.Logging;

public class GenerateApiToken
{
    private readonly ILogger<GenerateApiToken> _logger;

    public GenerateApiToken(ILogger<GenerateApiToken> logger)
    {
        _logger = logger;
    }

    [Function("GenerateApiToken")]
    public HttpResponseData Run([HttpTrigger(AuthorizationLevel.Function, "get")] HttpRequestData req)
    {
        _logger.LogInformation("开始生成API Token");

        // 从请求参数获取accountCode与API密钥
        var query = req.Query;
        string accountCode = query["accountCode"];
        string apiKey = query["apiKey"];

        // 生成dateToken
        long currentMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
        long ttl = 31536000000;
        long dateToken = currentMs + ttl;

        // 构造预请求路径
        string apiPath = $"/{accountCode}/data?fromDate=last6Hours&granularity=minute&type=ALL%2CVOD%2CLIVE&operation=reduceJoin&metrics=bufferratio";
        string preUrl = $"{apiPath}&dateToken={dateToken}";

        // 生成MD5 Token
        using MD5 md5 = MD5.Create();
        byte[] hashBytes = md5.ComputeHash(Encoding.UTF8.GetBytes(preUrl + apiKey));
        string token = BitConverter.ToString(hashBytes).Replace("-", "").ToLowerInvariant();

        // 拼接完整请求URL
        string baseUrl = "https://api.xxxxx.com";
        string fullRequestUrl = $"{baseUrl}{preUrl}&token={token}";

        // 返回结果
        var response = req.CreateResponse(HttpStatusCode.OK);
        response.WriteString(fullRequestUrl);
        return response;
    }
}

步骤2：在ADF中调用Function并获取数据

1. 添加第一个Web活动：
   • 设置URL为Azure Function的触发地址，方法为GET。
   • 添加查询参数：accountCode（管道参数）、apiKey（从Azure Key Vault获取，使用@secret('KeyVaultLinkedService', 'ApiKeySecret')）。
   • 将活动输出保存到管道变量fullRequestUrl，表达式为@activity('调用Token生成Function').output。
2. 添加第二个Web活动：
   • 设置URL为@variables('fullRequestUrl')，方法为GET，获取API数据。
3. 后续可添加Copy活动将API响应数据写入目标数据源（如Azure SQL、ADLS等）。

安全最佳实践 #

• 密钥管理：所有敏感信息（如API密钥）必须存储在Azure Key Vault，ADF通过Linked Service引用，避免明文暴露。
• 权限控制：确保Integration Runtime、Azure Function拥有访问目标API、存储资源的最小权限。

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