Azure门户中Python版Function App实现Azure SQL MI数据读写转换

整体代码结构（门户内） #

在Azure门户的Function App中，每个函数对应一个独立文件夹，核心文件包括：

• __init__.py：Python业务逻辑代码（数据读取、转换、写入）
• function.json：输入/输出绑定配置（定义SQL连接、读写语句）
• 根目录的requirements.txt：声明依赖包（自动安装）

1. 配置输入/输出绑定（function.json） #

通过绑定直接关联SQL MI，无需手动管理连接池。以下是支持多表读取+单表写入的配置示例：

{
  "scriptFile": "__init__.py",
  "bindings": [
    // 读取第一张表
    {
      "name": "inputTable1",
      "type": "sql",
      "direction": "in",
      "commandText": "SELECT * FROM SourceTable1",
      "connectionStringSetting": "SqlMiConnStr"
    },
    // 读取第二张表
    {
      "name": "inputTable2",
      "type": "sql",
      "direction": "in",
      "commandText": "SELECT * FROM SourceTable2",
      "connectionStringSetting": "SqlMiConnStr"
    },
    // 写入目标表
    {
      "name": "outputTargetTable",
      "type": "sql",
      "direction": "out",
      "commandText": "INSERT INTO TargetTable (Col1, Col2, NewCalculatedCol) VALUES (@Col1, @Col2, @NewCalculatedCol)",
      "connectionStringSetting": "SqlMiConnStr"
    }
  ]
}

• connectionStringSetting：对应Function App应用设置中的连接字符串名称，避免硬编码
• 输入绑定的commandText支持任意合法SQL查询（关联、过滤均可）
• 输出绑定的commandText需与后续代码中传入的字段名对应（@开头为参数占位符）

2. 核心业务代码（init.py） #

实现多表读取、数据转换、批量写入逻辑：

import azure.functions as func
import pandas as pd

def main(
    inputTable1: func.SqlRowList,
    inputTable2: func.SqlRowList,
    outputTargetTable: func.Out[func.SqlRowList]
) -> None:
    # 1. 将SQL输入绑定的行数据转换为DataFrame
    df_source1 = pd.DataFrame([dict(row) for row in inputTable1])
    df_source2 = pd.DataFrame([dict(row) for row in inputTable2])

    # 2. 数据转换示例：关联两张表+新增计算列
    df_merged = pd.merge(df_source1, df_source2, on="CommonKey", how="inner")
    df_merged["NewCalculatedCol"] = df_merged["NumCol1"] * df_merged["NumCol2"]  # 自定义计算逻辑

    # 3. 转换为SQL绑定支持的格式，批量写入目标表
    output_rows = []
    for _, row in df_merged.iterrows():
        output_rows.append(func.SqlRow({
            "Col1": row["SourceCol1"],
            "Col2": row["SourceCol2"],
            "NewCalculatedCol": row["NewCalculatedCol"]
        }))
    
    # 批量提交（比单条写入效率更高）
    outputTargetTable.set(func.SqlRowList(output_rows))

• func.SqlRowList：批量处理数据，减少数据库交互次数
• 转换逻辑可根据需求调整（如数据清洗、聚合、字段映射等）

3. 依赖声明（requirements.txt） #

在Function App根目录创建该文件，声明运行所需包：

azure-functions>=1.11.0
pandas>=2.0.0
pyodbc>=4.0.39

门户会自动安装这些依赖，无需手动部署。

4. 门户关键配置步骤 #

1. 添加SQL MI连接字符串：
   进入Function App → 「配置」→「应用程序设置」→「新建连接字符串」，名称设为SqlMiConnStr，值为SQL MI的连接字符串（示例：Server=tcp:<mi-name>.database.windows.net,1433;Database=<db-name>;Uid=<username>;Pwd=<password>;Encrypt=yes;Connection Timeout=30;）
2. 配置网络权限：
   确保SQL MI的防火墙规则允许Function App的出站IP访问（在Function App「概述」中可查看出站IP列表），或使用虚拟网络对等连接实现私有访问
3. 测试函数：
   在「代码+测试」页面点击「测试/运行」，查看日志输出验证读写是否成功

注意事项 #

• 如果需要更复杂的SQL操作（如动态查询），可直接在代码中使用pyodbc调用SQL MI，结合绑定实现混合模式
• 生产环境建议使用托管身份替代用户名密码，连接字符串改为Server=tcp:<mi-name>.database.windows.net,1433;Database=<db-name>;Authentication=Active Directory Managed Identity;，并给Function App的托管身份授予SQL MI的读写权限

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