Azure Persistence Features: Processing Lists

Question content

I have an azure persistent function written in python with a coordinator and two active functions

orchestrator calls the first active function and receives in return a list variable (the name list and this list can be dynamic each time the function is executed)

The next step is to call the second activity function for each list item (sequential processing - due to API limitations of the second activity function call)

#dynamically gets generated by the first activity function
payload=[1,2,3,4]            

tasks = [context.call_activity("secondfunction",ps) for ps in payload]
output = yield context.task_all(tasks)

I'm using something other than serial in my fanout method, but I can't seem to find an alternative to what I'm trying to do.

Also, in the host.json file, I tried to force that only one active function can run at a given time to avoid parallel processing

"extensions": {
    "durableTask": {
      "maxConcurrentActivityFunctions": 1,
      "maxConcurrentOrchestratorFunctions": 1
    }
  }

Also worth noting is that I can't pass the entire list to the activity function, as if I execute the activity function it will take more than 5-10 minutes, which is the timeout limit for azure functions, so trying to iterate over the list orchestration function

But the result is not continuous

Thank you very much for your feedback

Correct answer

You can try to use the following two methods to achieve your requirements:-

method 1:-

Myfunction_app.py:-

import azure.functions as func
import azure.durable_functions as df

myapp = df.dfapp(http_auth_level=func.authlevel.anonymous)

# http starter
@myapp.route(route="orchestrators/{functionname}")
@myapp.durable_client_input(client_name="client")
async def http_start(req: func.httprequest, client):
    function_name = req.route_params.get('functionname')
    instance_id = await client.start_new(function_name, none)  # pass the functionname here
    response = client.create_check_status_response(req, instance_id)
    return response

# orchestrator
@myapp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    cities = ["seattle", "tokyo", "london"]

    tasks = []
    for city in cities:
        tasks.append(context.call_activity("hello", city))

    # wait for all tasks to complete
    results = yield context.task_all(tasks)

    return results

# activity
@myapp.activity_trigger(input_name="city")
def hello(city: str):
    print(f"processing {city}...")
    # your activity function logic goes here
    result = f"hello {city}!"

    return result

Output: -

Function url:-

http://localhost:7071/api/orchestrators/hello_orchestrator

Method 2:-

function_app.py：-

import azure.functions as func
import azure.durable_functions as df

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# HTTP Starter
@myApp.route(route="orchestrators/{functionName}")
@myApp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client):
    function_name = req.route_params.get('functionName')
    instance_id = await client.start_new(function_name, None)  # Pass the functionName here
    response = client.create_check_status_response(req, instance_id)
    return response

# Orchestrator
@myApp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    # Call the first activity to get a list of names
    names_list = yield context.call_activity("get_names")

    # Process each name sequentially using the second activity
    results = []
    for name in names_list:
        result = yield context.call_activity("process_name", name)
        results.append(result)

    return results

# First Activity
@myApp.activity_trigger
def get_names():
    # Your logic to retrieve a dynamic list of names goes here
    # For demonstration purposes, returning a hardcoded list
    return ["John", "Alice", "Bob"]

# Second Activity
@myApp.activity_trigger(input_name="name")
def process_name(name: str):
    print(f"Processing {name}...")
    # Your logic to process each name goes here
    result = f"Hello {name}!"

    return result

The above is the detailed content of Azure Persistence Features: Processing Lists. For more information, please follow other related articles on the PHP Chinese website!