使用 Azure 函数 V2 Python（托管身份）将数据插入到 Azure SQL 中

Question

我正在遵循指南https://learn.microsoft.com/en-us/azure/azure-functions/functions-add-output-binding-azure-sql-vs-code?pivots=programming-language-蟒蛇，

我可以部署 Azure Function，但它不会将数据添加到 SQL 数据库，该数据库具有与之关联的 Azure 托管标识。

function_app.py

import azure.functions as func
import logging
import uuid
from azure.functions.decorators.core import DataType

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="hello", auth_level=func.AuthLevel.ANONYMOUS)
@app.generic_output_binding(arg_name="toDoItems", type="sql", CommandText="dbo.ToDo", ConnectionStringSetting="Server=X.database.windows.net; Authentication=Active Directory Default; Database=X;", data_type=DataType.STRING)
def test_function(req: func.HttpRequest, toDoItems: func.Out[func.SqlRow]) -> func.HttpResponse:
     logging.info('Python HTTP trigger function processed a request.')
     name = req.params.get('name')
     if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

     if name:
        toDoItems.set(func.SqlRow({"id": uuid.uuid4(), "title": name, "completed": "false", "url": "www"}))
        return func.HttpResponse(f"Hello {name}!")
     else:
        return func.HttpResponse(
                    "Please pass a name on the query string or in the request body",
                    status_code=400
                )

host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

Answer 1

我已使用以下步骤进行 SQL 输出绑定以在 python V2 函数中工作，这里我使用 ADO.NET（Microsoft Entra 无密码身份验证） 通过在函数应用程序中启用托管身份来使用 SQL 凭据。

1. 我对代码做了一些修改，它对我有用。

import azure.functions as func
import logging
from azure.functions.decorators.core import DataType
import uuid

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="hello", auth_level=func.AuthLevel.ANONYMOUS)
@app.generic_output_binding(arg_name="toDoItems", type="sql", CommandText="dbo.ToDo", ConnectionStringSetting="SqlConnectionString",
    data_type=DataType.STRING)
def test_function(req: func.HttpRequest, toDoItems: func.Out[func.SqlRow]) -> func.HttpResponse:
     logging.info('Python HTTP trigger function processed a request.')
     name = req.params.get('name')
     if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

     if name:
        item_id = str(uuid.uuid4())  # Convert UUID to string
        toDoItems.set(func.SqlRow({"id": item_id, "title": name, "completed": False, "url": "www"}))
        return func.HttpResponse(f"Hello {name}!")
     else:
        return func.HttpResponse(
                    "Please pass a name on the query string or in the request body",
                    status_code=400
                )

2. 我已在本地设置文件中添加了连接字符串。

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "SqlConnectionString": "Server=tcp:{serverName}.database.windows.net,1433;Initial Catalog={DatabaseName};Encrypt=True;TrustServerCertificate=False;Connection Timeout=30;Authentication=Active Directory Default;"
  }
}

• 请确保使用正确的连接字符串，否则您也可以从下面复制并粘贴它

3. 在本地执行该函数时，我能够将行插入数据库中。

• 请注意，我已通过导航至服务器 -> 网络在防火墙规则中添加本地客户端 IP 地址以进行本地开发。

4. 我在应用程序设置中添加了 AzureWebJobsFeatureFlagsthe 和 SqlConnectionString，如下所示 -

5. 然后，我将函数部署到函数应用程序中，并且部署成功。

• 要允许函数应用程序访问数据库，您需要在服务器中启用以下异常。

6. 然后我在函数应用程序中启用了托管身份。

• 我在数据库中执行了以下脚本，以便向托管身份授予权限。

CREATE USER [managed_identity_name] FROM EXTERNAL PROVIDER;
ALTER ROLE db_datareader ADD MEMBER [managed_identity_name];
ALTER ROLE db_datawriter ADD MEMBER [managed_identity_name];
ALTER ROLE db_ddladmin ADD MEMBER [managed_identity_name];
GO

7. 执行4-6步后，即可在函数app中成功执行该函数。