Azure function V2 风格编码 - Http 触发器和 blob 输出绑定

Question

我有以下代码，在本地启动时可以完美运行。我输入 URL，该函数通过从 API 检索 TFL 负载并将数据存储在 Blob 存储中来执行其任务。这个过程是通过访问 URL http://localhost:7071/api/tfl_blob 来触发的。

尽管如此，将此代码部署到 Azure 平台后，部署继续进行，没有任何错误消息，但该函数不会出现在函数应用程序中。虽然我可以访问应用程序文件，但触发它时遇到困难，并且当我导航到功能概述页面时，该功能明显缺失。

可能是什么问题以及如何解决？

功能代码：

import datetime
import json
import requests
import azure.functions as func
import logging
import uuid

app = func.FunctionApp()


def generate_blob_path():
    current_datetime = datetime.datetime.now()
    date_str = current_datetime.strftime('%Y-%m-%d')
    unique_id = str(uuid.uuid4())
    blob_path = f"bloboutput/{date_str}/{unique_id}.json"
    return blob_path


_blob_path = generate_blob_path()


@app.function_name(name="HttpTflBlobOut")
@app.route(route="tfl_blob")
@app.blob_output(arg_name="blob",
                 path=_blob_path,
                 connection="AzureWebJobsStorage")
def main(req: func.HttpRequest, blob: func.Out[str]) -> func.HttpResponse:
    try:
        result = requests.get('https://api.tfl.gov.uk/line/mode/tube/status',
                              headers={"content-type": "application/json", "charset": "utf-8"})

        logging.info('Python HTTP trigger function processed the API request to TFL')

        since_id = None
        timeline = json.loads(result.text)

        tflLineStatus = []
        for t in timeline:
            if since_id is None:
                since_id = t["id"]

            tflLineStatus.append({
                "linename": t["id"],
                "linestatus": t["lineStatuses"][0]["statusSeverityDescription"],
                "timestamp": t["created"]
            })

        if since_id is None:
            since_id = req.state.since_id

        ans = {
            "state": {
                since_id: since_id
            },
            "schema": {
                "tflLineStatus": {
                    "primary_key": ["linename"]
                }
            },
            "insert": {
                "tflLineStatus": tflLineStatus
            },
            "hasMore": False
        }

        logging.info('Payload is constructed')

        payload = json.dumps(ans, indent=4)
        logging.info('Payload is converted to a string')
        blob.set(payload)

        return func.HttpResponse(f"Function executed successfully", status_code=200)
    except Exception as e:
        logging.error(f"Error: {str(e)}")
        return func.HttpResponse("Internal Server Error", status_code=500)

主机.json

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

local.settings.json：

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "AzureWebJobsStorage": <match with storage account connection string and also set on the az function>,
    "FUNCTIONS_EXTENSION_VERSION": "~4",
    "APPLICATIONINSIGHTS_CONNECTION_STRING": "<match with the function config on az>"
  },
  "ConnectionStrings": {}
}

我的期望是我应该能够使用以下 URL 格式通过 HTTP 请求调用该函数：

https://<function app name>.azurewebsites.net/api/<function name>?code=<function key>

。它的执行方式应该与本地的执行方式类似。但是，我遇到了 HTTP ERROR 404，表明无法找到该页面。

Answer 1

我在我的环境中尝试了你的代码。我能够运行函数代码并部署它。

代码：

import datetime
import json 
import requests 
import azure.functions as func 
import logging 
import uuid

app = func.FunctionApp()

def generate_blob_path():
    current_datetime = datetime.datetime.now()
    date_str = current_datetime.strftime('%Y-%m-%d')
    unique_id = str(uuid.uuid4())
    blob_path = f"bloboutput/{date_str}/{unique_id}.json"
    return blob_path

_blob_path = generate_blob_path()

@app.function_name(name="HttpTflBlobOut") @app.route(route="tfl_blob")
@app.blob_output(arg_name="blob",
                 path=_blob_path,
                 connection="AzureWebJobsStorage") def main(req: func.HttpRequest, blob: func.Out[str]) -> func.HttpResponse:
    try:
        result = requests.get('https://api.tfl.gov.uk/line/mode/tube/status',
                              headers={"content-type": "application/json", "charset": "utf-8"})

        logging.info('Python HTTP trigger function processed the API request to TFL')

        since_id = None
        timeline = json.loads(result.text)

        tflLineStatus = []
        for t in timeline:
            if since_id is None:
                since_id = t["id"]

            tflLineStatus.append({
                "linename": t["id"],
                "linestatus": t["lineStatuses"][0]["statusSeverityDescription"],
                "timestamp": t["created"]
            })

        if since_id is None:
            since_id = req.state.since_id

        ans = {
            "state": {
                since_id: since_id
            },
            "schema": {
                "tflLineStatus": {
                    "primary_key": ["linename"]
                }
            },
            "insert": {
                "tflLineStatus": tflLineStatus
            },
            "hasMore": False
        }

        logging.info('Payload is constructed')

        payload = json.dumps(ans, indent=4)
        logging.info('Payload is converted to a string')
        blob.set(payload)

        return func.HttpResponse(f"Function executed successfully", status_code=200)
    except Exception as e:
        logging.error(f"Error: {str(e)}")
        return func.HttpResponse("Internal Server Error", status_code=500) ```

local.settings.json：

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "AzureWebJobsStorage": "<storage_connec_string>",
    "FUNCTIONS_EXTENSION_VERSION": "~4",
    "APPLICATIONINSIGHTS_CONNECTION_STRING": "<applicationInsights_connec_string>"
  },
  "ConnectionStrings": {}
}

需求.txt：