How to use queue triggers in Azure Durable Python Functions with managed identity

Question

I'm developing a queue trigger for an Azure Durable Function. When using the "queue_trigger" decorator, I'm required to provide a connection string. However, I need to avoid using a connection string directly and instead use Managed Identity to connect to the Storage Account/Queue Storage, ideally with DefaultAzureCredential() or a similar method. In the example below, I currently have the "QueueConnectionString" set up in the environment variables, but I want to replace this with Managed Identity for secure access to the queue.

import azure.functions as func
import logging
import azure.durable_functions as adf

myApp = adf.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@myApp.durable_client_input(client_name="client")
@myApp.queue_trigger(arg_name="azqueue", queue_name="test", connection="QueueConnectionString")
async def begin_data_entry(azqueue: func.QueueMessage, client):
    logging.info('Python HTTP trigger function processed a request.')
    await client.start_new("activity_function_name", client_input={})

You can this check this process for Manged Identity in functions. — Dasari Kamali
– Dasari Kamali, Commented Nov 8, 2024 at 5:29
@DasariKamali Is there a way to set up "QueueConnectionString__queueServiceUri" without using the application settings environment variable section? Since I am using Azure App Configuration to store and access environment variables, I tried hardcoding the URL in the code as an environment variable, like below. However, it didn’t work for me. os.environ["QueueConnectionString__queueServiceUri"] = "https://########.queue.core.windows.net/" — user25648715
– user25648715, Commented Nov 8, 2024 at 9:16

Dasari Kamali · Accepted Answer · 2024-11-08 14:43:56Z

I created a sample queue trigger for an Azure Durable Function with Managed identity using DefaultAzureCredential to process a message to the Azure Storage Queue and it worked fine for me.

It's better to add the storage URI QueueConnectionString__queueServiceUri in the local.settings.json as below.

local.settings.json :

{
    "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "QueueConnection__queueServiceUri": "https://<storage_name>.queue.core.windows.net/"
  }
}

Add the below connection in the code, it will load the storage URI from local.settings.json.

connection="QueueConnection__queueServiceUri"

Code :

import azure.functions as func
import logging
import azure.durable_functions as adf
from azure.identity import DefaultAzureCredential
from azure.storage.queue import QueueClient

myApp = adf.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@myApp.durable_client_input(client_name="client")
@myApp.queue_trigger(arg_name="azqueue", queue_name="test", connection="QueueConnection__queueServiceUri")
async def begin_data_entry(azqueue: func.QueueMessage, client):
    logging.info('Python HTTP trigger function processed a request.')
    await client.start_new("activity_function_name", client_input={})

I have added the owner role to the service principle and Storage Queue Data Contributor role to the function app in the Azure Storage account as shown below.

enter image description here

Make sure to add the below URI to the Azure Function App > Environment Variables > App settings as shown below.

"QueueConnection__queueServiceUri": "https://<storage_name>.queue.core.windows.net/"

enter image description here

Collectives™ on Stack Overflow

How to use queue triggers in Azure Durable Python Functions with managed identity

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related