Integrating Azure SQL with Durable Activity Function Using SQL Binding in Python Programming Model V2

Question

Hello,

I'm working with the Python V2 programming model for Azure Functions and attempting to use a SQL input binding within a durable activity function. I want to pass a parameter to the activity function and use it as a variable in the SQL query string.

However, in my current implementation, the activity function returns a blank row. I have verified the following:

• The SQL query works as expected when tested directly in SSMS
• The durable function and SqlConnectionString are correctly configured — when I hardcode the parameter value directly into the query string, the function returns the correct results
• The same query and binding setup also works when tested in a regular (non-durable) Azure Function, where I pass the parameter via the function route

These checks suggest the issue may lie in how the binding handles dynamic parameters in the context of a durable activity function.

I came across a similar question on Microsoft Q&A that appears to support my approach.

I'd appreciate any advice or suggestions for resolving this issue.

Thank you!

Example durable function:

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

bp = df.Blueprint()

@bp.route(route="startOrchestrator")
@bp.durable_client_input(client_name="client")
async def start_orchestrator(req: func.HttpRequest, client):
    instance_id = await client.start_new("my_orchestrator")

    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)


@bp.orchestration_trigger(context_name="context", orchestration="my_orchestrator")
def my_orchestrator(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('lot_info', 'lot_name_1')
    result2 = yield context.call_activity('lot_info', 'lot_name_2')
    return [result1, result2]


@bp.activity_trigger(input_name="lot")
@bp.sql_input(arg_name="lotInfo", command_text="SELECT Lot, Program, OrderProductID, ProductID, LotType FROM wip.LotDetail WHERE Lot=@LOT", command_type="Text", parameters="@LOT={lot}", connection_string_setting="SqlConnectionString")
def lot_info(lot: str, lotInfo: func.SqlRowList) -> list:
    rows = list(map(lambda r: json.loads(r.to_json()), lotInfo))
    return rows

requirements.txt

azure-functions
azure-functions-durable>=1.2.4
azure-identity
numpy==2.2.4
pandas==2.2.3
requests==2.32.3

host.json

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

function_app.py

import azure.functions as func
import FILM 
import IVS 
import Durable 

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

app.register_functions(FILM.bp)
app.register_functions(IVS.bp)
app.register_functions(Durable.bp)

Answer 1

Hi @Gaia Dennison ,

I have used below code and approach which works for me:

Structure:

project-root/

│

├── __pycache__/              

├── .venv/                    

├── .vscode/                  

├── .funcignore               

├── .gitignore                

├── function_app.py           

├── host.json                 

├── local.settings.json       

├── requirements.txt          

├── testchain.py     

function_app.py:

import azure.functions as func

import azure.durable_functions

from testchain import cho

rith = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

rith.register_blueprint(cho)

Used register_blueprint in registering it.

testchain. py:

import logging as ri_lg

import azure.functions as func

import azure.durable_functions as df

import json

cho = df.Blueprint()

@cho.route(route="startOrchestrator")

@cho.durable_client_input(client_name="client")

async def start_orchestrator(req: func.HttpRequest, client):

    instance_id = await client.start_new("my_orchestrator")

    ri_lg.info(f"Hello Rithwik, Started orchestration with ID = '{instance_id}'.")

    return client.create_check_status_response(req, instance_id)

@cho.orchestration_trigger(context_name="context", orchestration="my_orchestrator")

def my_orchestrator(context: df.DurableOrchestrationContext):

    result1 = yield context.call_activity('rith_cho', 'rith_cho_1')

    result2 = yield context.call_activity('rith_cho', 'rith_cho_2')

    return [result1, result2]

@cho.activity_trigger(input_name="rb")

@cho.sql_input(arg_name="rbch", command_text="SELECT * FROM [dbo].[test]", command_type="Text", parameters="@LOT={rb}", connection_string_setting="Testcon")

def rith_cho(rb: str, rbch: func.SqlRowList) -> list:

    rows = list(map(lambda r: json.loads(r.to_json()), rbch))

    return rows

local.settings.json:

{

  "IsEncrypted": false,

  "Values": {

    "AzureWebJobsStorage": "DefaultEndpointsProtocol=https;AccountName=rithwik953b;AccountKey=JDfotYrithwikchotuwQ==;EndpointSuffix=core.windows.net",

    "FUNCTIONS_WORKER_RUNTIME": "python",

    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",

    "Testcon":"Server=tcp:rithwik.database.windows.net,1433;Initial Catalog=test1;Persist Security Info=False;User ID=rithwik;Password=password;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=False;Connection Timeout=30;",

    "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"

  }

}

host.json:

{

  "version": "2.0",

  "logging": {

    "applicationInsights": {

      "samplingSettings": {

        "isEnabled": true,

        "excludedTypes": "Request"

      }

    }

  },

  "extensionBundle": {

    "id": "Microsoft.Azure.Functions.ExtensionBundle",

    "version": "[4.*, 5.0.0)"

  }

}

In SQL :

Output:

Try to follow correct structure and give correct connection string value, it will work as it worked for me.

---

If this answer was helpful, please click "Accept the answer" and mark Yes, as this can help other community members.

If you have any other questions or are still experiencing issues, feel free to ask in the "comments" section, and I'd be happy to help.