如何使用Robot Framework连接Azure Data Factory中的Linked Service并执行查询

前置准备 #

• 安装依赖库：

  • 执行pip install robotframework-requests安装RequestsLibrary，用于调用Azure REST API
  • 针对目标数据源安装对应库，比如SQL Server场景执行pip install robotframework-databaselibrary pyodbc

• 准备必要信息：

  • 服务主体（Service Principal）的租户ID、客户端ID、客户端密钥
  • ADF实例的资源组名称、工厂名称
  • 目标Linked Service的名称

步骤1：获取Azure AD访问令牌 #

调用ADF的REST API需要先获取有效访问令牌，Robot Framework代码示例如下：

*** Settings ***
Library    RequestsLibrary
Library    Collections

*** Variables ***
${TENANT_ID}    你的租户ID
${CLIENT_ID}    你的客户端ID
${CLIENT_SECRET}    你的客户端密钥
${TOKEN_URL}    https://login.microsoftonline.com/${TENANT_ID}/oauth2/v2.0/token
${ADF_RESOURCE}    https://management.azure.com/

*** Test Cases ***
Get Azure AD Access Token
    Create Session    azure_auth    ${TOKEN_URL}
    &{payload}    Create Dictionary    grant_type=client_credentials    client_id=${CLIENT_ID}    client_secret=${CLIENT_SECRET}    scope=${ADF_RESOURCE}/.default
    ${response}    Post Request    azure_auth    ${TOKEN_URL}    data=${payload}
    ${token}    Get From Dictionary    ${response.json()}    access_token
    Set Suite Variable    ${ACCESS_TOKEN}    ${token}
    Log    已获取访问令牌

步骤2：从ADF提取Linked Service的连接信息 #

通过ADF Management API查询目标Linked Service的配置，提取数据源连接参数（以Azure SQL Database的Linked Service为例）：

*** Variables ***
${ADF_API_URL}    https://management.azure.com/subscriptions/你的订阅ID/resourceGroups/${RESOURCE_GROUP}/providers/Microsoft.DataFactory/factories/${ADF_NAME}/linkedservices/${LINKED_SERVICE_NAME}?api-version=2018-06-01

*** Test Cases ***
Fetch Linked Service Connection Details
    Create Session    adf_api    ${ADF_API_URL}
    &{headers}    Create Dictionary    Authorization=Bearer ${ACCESS_TOKEN}    Content-Type=application/json
    ${response}    Get Request    adf_api    ${ADF_API_URL}    headers=${headers}
    # 根据Linked Service类型调整参数提取路径，此处为Azure SQL的连接字符串
    ${conn_string}    Get From Dictionary    ${response.json()}[properties][typeProperties]    connectionString
    Set Suite Variable    ${DB_CONN_STRING}    ${conn_string}
    Log    已获取数据源连接字符串

注意：不同类型的Linked Service，typeProperties下的参数结构不同。比如Blob Storage的Linked Service会直接返回connectionString，而部分数据源可能拆分存储server、database等字段，需根据实际情况调整提取逻辑。

步骤3：连接数据源并执行查询 #

使用DatabaseLibrary连接到数据源，执行查询操作：

*** Settings ***
Library    DatabaseLibrary

*** Test Cases ***
Execute Query via Linked Service
    # 初始化数据库连接（以SQL Server的pyodbc驱动为例）
    Connect To Database Using Custom Params    pyodbc    ${DB_CONN_STRING}
    # 执行目标查询
    ${result}    Query    SELECT TOP 10 * FROM 你的目标表名
    # 输出或处理查询结果
    Log    查询结果：${result}
    # 关闭数据库连接
    Disconnect From Database

注意事项 #

• 确保服务主体拥有ADF的Data Factory Contributor或Reader权限，同时具备目标数据源的访问权限（比如SQL Server的db_reader角色）
• 若Linked Service使用Azure Key Vault存储敏感信息（如密码），需额外调用Key Vault API获取密钥，替换连接字符串中的占位符
• 不同数据源的连接驱动需对应调整，比如MySQL用mysql-connector-python，PostgreSQL用psycopg2

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