Azure Function App 用 Python 而非 SQL 处理数据

2026-11-04技术老齐

可以在同一个 Azure Function 中完成读取、Python 转换和写入，不必拆成”读取函数、转换函数、写入函数”，也不必通过 HTTP 在函数之间传递表数据。

在 Python v2 编程模型中，同一个函数可以配置一个或多个 sql_input，以及一个 sql_output。输入绑定将查询结果注入函数参数，转换后的数据再通过 func.Out 交给输出绑定。

准备目标表

下面的表结构用于说明写入方式。实际使用时，需要调整列名、数据类型和主键：

CREATE TABLE dbo.ToDoResult
(
    Id uniqueidentifier NOT NULL PRIMARY KEY,
    title nvarchar(200) NULL,
    completed bit NULL,
    ProcessedAt datetime2(7) NOT NULL
);

SQL output binding 会根据目标表的主键执行写入或更新，也就是 upsert。目标表需要有主键，输出对象的字段名也要与目标表列名对应。

如果目标表包含 identity、计算列、触发器或其他特殊约束，建议先用少量数据测试写入行为。

配置数据库连接

在 Function App 的应用设置中创建：

SqlConnectionString

将它的值设为 Azure SQL MI 的连接字符串。用户名、密码和连接字符串不要直接写在 Python 代码中。

使用 Function App 托管身份时，需要完成以下配置：

为 Function App 启用托管身份。
在 Azure SQL MI 中为该身份创建用户，并授予源表读取权限和目标表写入权限。

数据库管理员可以根据实际身份和权限模型执行类似配置：

CREATE USER [<function-app-identity>] FROM EXTERNAL PROVIDER;

GRANT SELECT ON dbo.ToDo TO [<function-app-identity>];
GRANT SELECT, INSERT, UPDATE ON dbo.ToDoResult TO [<function-app-identity>];

通过 Microsoft Entra 身份连接 SQL MI 时，可能还需要配置 SQL MI 的 Entra 管理员。连接字符串的具体格式取决于使用系统分配身份、用户分配身份还是 SQL 身份验证。

如果 SQL MI 只能通过私有网络访问，Function App 还需要具备访问该网络的出站路径，相关配置通常包括 VNet Integration、路由、DNS 和网络安全规则。代码配置完成后，还要单独确认网络是否连通。

在一个函数中读取、转换并写入

下面继续使用已有的 HTTP 触发器。示例从 dbo.ToDo 读取数据，增加 ProcessedAt 列，再写入 dbo.ToDoResult：

import json
from datetime import datetime, timezone

import azure.functions as func

app = func.FunctionApp()


@app.function_name(name="ProcessToDo")
@app.route(route="processtodo", methods=["POST"])
@app.sql_input(
    arg_name="todo_rows",
    command_text="""
        SELECT Id, title, completed
        FROM dbo.ToDo
    """,
    command_type="Text",
    connection_string_setting="SqlConnectionString"
)
@app.sql_output(
    arg_name="result_rows",
    command_text="dbo.ToDoResult",
    connection_string_setting="SqlConnectionString"
)
def process_todo(
    req: func.HttpRequest,
    todo_rows: func.SqlRowList,
    result_rows: func.Out[func.SqlRowList]
) -> func.HttpResponse:
    source_rows = [
        json.loads(row.to_json())
        for row in todo_rows
    ]

    processed_at = datetime.now(timezone.utc).isoformat()

    output_rows = [
        func.SqlRow.from_dict({
            "Id": row["Id"],
            "title": row.get("title"),
            "completed": row.get("completed"),
            "ProcessedAt": processed_at
        })
        for row in source_rows
    ]

    result_rows.set(func.SqlRowList(output_rows))

    return func.HttpResponse(
        json.dumps({
            "readCount": len(source_rows),
            "writeCount": len(output_rows)
        }),
        status_code=200,
        mimetype="application/json"
    )

需要注意以下几点：

todo_rows 是 input binding 注入的查询结果。
result_rows 是 output binding 参数，输入数据无需先通过 HTTP 响应返回。
command_text="dbo.ToDoResult" 指向目标表，其中填写的不是 INSERT 语句。
func.SqlRow.from_dict() 将 Python 字典转换为 SQL output binding 可以处理的行。
字典中的 Id、title、completed 和 ProcessedAt 必须与目标表列名一致。
代码使用稳定的 Id 作为主键。再次处理同一条数据时，output binding 会按主键更新对应行，不会直接追加重复记录。

datetime.now(timezone.utc).isoformat() 会生成 ISO 8601 字符串。如果当前 SQL binding 或目标列不能按预期转换该值，可以改用目标数据库明确接受的日期时间格式，并在实际运行环境中测试。

同时读取两张表

同一个函数可以声明多个 input binding，每个 binding 对应一个函数参数：

@app.function_name(name="ProcessTwoTables")
@app.route(route="processtwotables", methods=["POST"])
@app.sql_input(
    arg_name="customers",
    command_text="SELECT CustomerId, Name FROM dbo.Customer",
    command_type="Text",
    connection_string_setting="SqlConnectionString"
)
@app.sql_input(
    arg_name="orders",
    command_text="SELECT OrderId, CustomerId, Amount FROM dbo.CustomerOrder",
    command_type="Text",
    connection_string_setting="SqlConnectionString"
)
@app.sql_output(
    arg_name="results",
    command_text="dbo.CustomerOrderResult",
    connection_string_setting="SqlConnectionString"
)
def process_two_tables(
    req: func.HttpRequest,
    customers: func.SqlRowList,
    orders: func.SqlRowList,
    results: func.Out[func.SqlRowList]
) -> func.HttpResponse:
    customer_rows = [
        json.loads(row.to_json())
        for row in customers
    ]
    order_rows = [
        json.loads(row.to_json())
        for row in orders
    ]

    customer_by_id = {
        row["CustomerId"]: row
        for row in customer_rows
    }

    transformed = []

    for order in order_rows:
        customer = customer_by_id.get(order["CustomerId"])

        transformed.append(
            func.SqlRow.from_dict({
                "OrderId": order["OrderId"],
                "CustomerId": order["CustomerId"],
                "CustomerName": customer["Name"] if customer else None,
                "Amount": order["Amount"]
            })
        )

    results.set(func.SqlRowList(transformed))

    return func.HttpResponse(
        json.dumps({"writeCount": len(transformed)}),
        status_code=200,
        mimetype="application/json"
    )

这段代码可以用来验证 Python 是否能够完成数据合并和转换。使用时需要替换真实的列名、数据类型和目标表主键。

什么时候需要拆成多个 Function

对于数据量较少、单次执行即可完成的 POC，将整个流程放在一个 Function 中最简单。遇到以下情况时，可以考虑拆分：

数据量较大，需要分批处理。
转换时间可能超过当前托管计划允许的函数执行时间。
读取、转换和写入需要分别重试。
数据来自多个系统，需要使用队列或 Durable Functions 编排处理过程。
各阶段需要独立扩缩容或独立监控。

函数之间不适合通过普通 HTTP 参数传递整张表。拆分后可以传递批次编号、文件地址或记录范围，再由各阶段通过 Queue、Blob、Data Lake 或 Durable Functions 协调。

Portal 开发的限制

如果 Function App 的 “Code + Test” 页面支持编辑当前 Python v2 项目，可以直接维护 function_app.py。Portal 的内联编辑能力受托管计划、操作系统和部署方式影响，并非所有 Python Function App 都能在其中完整编辑项目。

如果页面只读，或者无法管理项目文件，可以通过 Azure Cloud Shell 创建并部署项目。Cloud Shell 运行在 Azure 的浏览器环境中，无需在本地安装 VS Code。至于这种方式是否符合组织的安全要求，需要由内部安全团队确认。

已有的 SQL input binding 能够正常执行，通常说明 SQL binding 扩展已经加载。添加 sql_output 后，如果运行时提示找不到 binding，还需要检查 host.json 中的 extension bundle 配置，以及应用使用的 azure-functions 依赖版本。

运行时注意事项

SQL input binding 会先加载查询结果，再将其交给 Python。对大表直接执行 SELECT *，容易增加内存占用、执行时间和网络传输量。即使只是 POC，也建议：

明确指定列名，避免使用 SELECT *。
使用过滤条件或批次范围控制单次处理的数据量。
为连接字段和筛选字段建立合适的索引。
使用稳定主键，确保重复调用时可以安全执行 upsert。
不要在日志中输出连接字符串或完整的敏感数据。
先用少量记录验证字段类型，尤其是 datetime、decimal、二进制数据和空值。

input binding 的读取和 output binding 的写入通常不能视为一个由应用控制的数据库事务。如果业务要求读取与写入整体提交或整体回滚，需要使用数据库驱动显式管理连接和事务，不能只依赖这两个 binding。