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
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!