Request for Example on Integrating Azure SQL with Durable Activity Function Using SQL Binding in Python Programming Model V2

Question

Hello everyone.

I can't seem to find the answer either on Q&A or in the documentation. I

would like to connect to Azure SQL using a DURABLE ACTIVITY FUNCTION in Python programming model V2 via Azure Function SQL binding.

Could you please show me an example of how this should be correctly done?

There are examples for regular functions in model V2, but I haven't been able to do this for Durable. I don't know if it is even possible, or if it needs to be circumvented somehow (with some other Python library, not through function SQL binding)?

I tried the following method, but of course, there's a name collision between the returned arg_name from the SQL binding and the input_name argument passed to the activity:

@sqlconf.activity_trigger(activity="activity_name", input_name="input")
@sqlconf.sql_input(connection_string_setting="My_ConnStr", command_text="SELECT * FROM TableA", command_type="Text", arg_name="input")
def get_data(input: func.SqlRowList) -> func.HttpResponse:
    rows = list(map(lambda r: json.loads(r.to_json()), input))

    return func.HttpResponse(
        json.dumps(rows),
        status_code=200,
        mimetype="application/json"
    )

Answer 1

After several hours, I finally managed to succeed!

Until now, I thought that the function behind the activity trigger also had to have one parameter, but of course, I was mistaken. It's the

• Activity trigger that must have one,
• but the function can have more.

Linking the parameters from the activity trigger with arg_name from the SQL binding gave the expected result, and I was able to retrieve rows in activity from Azure SQL. I provide my example as a blueprint for all you guys who are looking to achieve similar results!

samples\durable\chaining\bp_sql_binding_example_df_chain.py

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

sql_binding_example_df_chain = df.Blueprint()

@sql_binding_example_df_chain.route(route="start_hello_cities_orchestrator_df_sql")
@sql_binding_example_df_chain.durable_client_input(client_name="client")
async def start_hello_cities_orchestrator_df_sql(req: func.HttpRequest, client):
    instance_id = await client.start_new("sql_binding_example_df_chain_orchestrator")
    
    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)

@sql_binding_example_df_chain.orchestration_trigger(context_name="context",orchestration="sql_binding_example_df_chain_orchestrator")
def sql_binding_example_df_chain_orchestrator(context: df.DurableOrchestrationContext):
    result0 = yield context.call_activity('say_hello_for_chain_sql', "Warszawa")
    result1 = yield context.call_activity('say_hello_for_chain_sql', "Tokyo")
    result2 = yield context.call_activity('say_hello_for_chain_sql', "Seattle")
    result3 = yield context.call_activity('say_hello_for_chain_sql', "London")
    return [result0, result1, result2, result3]

@sql_binding_example_df_chain.activity_trigger(input_name="city")
@sql_binding_example_df_chain.sql_input(arg_name="version", command_text="SELECT ver=@@VERSION, param=@City", command_type="Text", parameters="@City={city}", connection_string_setting="EM_ADF_FMWK_ConnStr")
def say_hello_for_chain_sql(city: str, version: func.SqlRowList) ->  list:
    rows = list(map(lambda r: json.loads(r.to_json()), version))
    return rows

function_app.py

import azure.functions as func
import logging

from samples.durable.chaining.bp_sql_binding_example_df_chain import sql_binding_example_df_chain

main = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

main.register_blueprint(sql_binding_example_df_chain)

Execution log:

Execution results: