Azure Function - 参数在 Python 中声明，但不在 function.json 中声明：{'starter'}

Question

我有一个带有 Azure 函数的项目，该函数有两个 http 触发器函数。

import azure.functions as func
from func_utils.main_func import http_trigger_function
from func_utils.get_data_trigger import main_second_trigger_func


#Instantiate the HTTP Function app 
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.route(route="http_trigger_function")
def http_trigger_func(req: func.HttpRequest, context) -> func.HttpResponse: 

    return http_trigger_function(req, context)


# Segundo HTTP Trigger
@app.route(route="get-data")
def main_second_trigger(req: func.HttpRequest) -> func.HttpResponse:

    return main_second_trigger_func(req)

我想使用持久功能并做这样的事情：

def http_trigger(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    """HTTP trigger to start the orchestration."""
    client = df.DurableOrchestrationClient(starter)
    order_details = req.get_json()

    # Start the orchestration
    instance_id = client.start_new("orchestrator_function", None, order_details)

    return func.HttpResponse(f"Orchestration started with ID = {instance_id}")

我只是像那个例子一样在我的函数中声明了启动器，此刻没有使用启动器只是为了看看它是如何工作的。但是当我尝试启动该功能时遇到此错误：

Worker failed to index functions
[2024-11-28T23:08:32.196Z] Result: Failure
Exception: FunctionLoadError: cannot load the http_trigger_func function: the following parameters are declared in Python but not in function.json: {'starter'}     
Stack:   File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\dispatcher.py", line 413, in _handle__functions_metadata_request
    self.load_function_metadata(
  File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\dispatcher.py", line 393, in load_function_metadata
    self.index_functions(function_path, function_app_directory)) \
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\dispatcher.py", line 773, in index_functions   
    loader.process_indexed_function(
  File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\loader.py", line 139, in process_indexed_function
    function_info = functions_registry.add_indexed_function(
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\functions.py", line 450, in add_indexed_function
    deferred_bindings_enabled) = self.validate_function_params(
                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11\WINDOWS\X64\azure_functions_worker\functions.py", line 137, in validate_function_params
    raise FunctionLoadError(

这是我的host.json：

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": {
    "http": {
      "routePrefix": "api",
      "hsts": {
        "isEnabled": true,
        "includeSubDomains": true,
        "maxAge": "365"
      },
      "customHeaders": {
        "X-Content-Type-Options": "nosniff",
        "X-Frame-Options": "DENY",
        "X-XSS-Protection": "1; mode=block",
        "Content-Security-Policy": "default-src 'self'",
        "X-Content-Security-Policy": "default-src 'self'"
      }
    },
    "durableTask": {
      "hubName": "DurableFunctionsHub",
      "maxConcurrentActivityFunctions": 10,
      "maxConcurrentOrchestratorFunctions": 10
      }
  }
}

我没有 function.json

我尝试按如下方式声明启动器：

starter: df.DurableOrchestrationClient

但是没用

Answer 1

要启动持久功能，您不需要启动器，但您可以在传递 URL 时使用参数中的函数名称并遵循我的SO-Answer：

import azure.functions as ric
import logging
import json
import azure.durable_functions as ch

rithapp=ch.DFApp(http_auth_level=ric.AuthLevel.ANONYMOUS)
@rithapp.route(route="http_trigger")
def http_trigger(req: ric.HttpRequest) -> ric.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    return ric.HttpResponse(
             "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

@rithapp.route(route="orchestrators/{functionName}")
@rithapp.durable_client_input(client_name="client")
async def http_start(req: ric.HttpRequest, client):
    function_name = req.route_params.get('functionName')
    instance_id = await client.start_new(function_name)
    response = client.create_check_status_response(req, instance_id)
    return response

@rithapp.orchestration_trigger(context_name="context")
def rith_orchestrator(context):
    
    rith_act_data = {"testdata": "Rithwik"}
    rith_serialised_act_data = json.dumps(rith_act_data)
    logging.info(f"Activity result: {rith_serialised_act_data}")
    rith_result = yield context.call_activity("hello_rithwik", rith_serialised_act_data)
    return rith_result

@rithapp.activity_trigger(input_name="rithjson")
def hello_rithwik(rithjson: str):
    json_activity_data = json.loads(rithjson)
    data = json_activity_data["testdata"] 
    logging.info("Helo Rithwik, Executing Activity Trigger")
    return data

因此，要调用orchestrator，请使用：

http://localhost:7071/api/orchestrators/rith_orchestrator

这是持久函数的格式，您可以从http启动调用orchestrator，然后调用活动触发器。所以你还必须有协调器。

输出：

调用http_start：

我没有 function.json

您没有 function.json，当您在本地创建 azure 函数时，根据 Microsoft-Document 其结构将是：

<Rth Folder>/
 | - .venv/
 | - .vscode/
 | - function_app.py
 | - extra_rith_funs.py
 | - tests/
 | | - test_rith_func.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile