Python Azure Function, queue-triggered alternative `connection` for `function.json`?

Is there a way to use a Managed Identity (either System or User-assigned) for authenticating a Function and Storage Account?

Example

• In a Python Azure Function's function.json file, there is a bindings object:

"bindings": [
    {
      "name": "msg",
      "type": "queueTrigger",
      "direction": "in",
      "queueName": "<my-queue-name>",
      "connection": "CONN_STRING"
    }
  ]

• The connection parameter refers to a value in local.settings.json when run locally OR an App Setting when deployed to Azure.
• These can be either hardcoded in these areas or use a Key Vault Secret Reference

Question:

• Instead of using a Connection String (whether KV Reference or not), can I can grant the Functions Managed Identity a Storage Queue Contributor RBAC role?
• I don't want to handle these Connection Strings

Is this possible? If so, what value do I put in the connection parameter?

• Instead of using a Connection String (whether KV Reference or not), can I can grant the Functions Managed Identity a Storage Queue Contributor RBAC role?

Instead of passing the Storage Account's connection string, you can achieve the same with an Identity-Based connection. Refer MSDOC:

You can use AzureWebJobsStorage__queueServiceUri setting to achieve your requirement.

• You must have Storage Queue Data Contributor role.

• Assign Storage Queue Data Reader and Storage Queue Data Message Processor roles in the Storage Account.
• Enable Managed Identity in the Function App.
• Set the Application Setting in the local.setting.json AzureWebJobsStorage__accountName to the name of the your storage account as its value.

local.settings.json:

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "AzureWebJobsStorage__queueServiceUri": "https://<your_storage_account_name>.queue.core.windows.net/"
  }
}

function_app.py:

app = func.FunctionApp()


@app.queue_trigger(arg_name="azqueue", queue_name="<queuename>",
                               connection="AzureWebJobsStorage") 
def queue_trigger(azqueue: func.QueueMessage):
    logging.info('Python Queue trigger processed a message: %s',
                azqueue.get_body().decode('utf-8'))

Local Response:

(.venv) C:\Users\username\pythonazfn>func host start
Found Python version 3.11.6 (py).

Azure Functions Core Tools
Core Tools Version:       4.0.5455 Commit hash: N/A  (64-bit)
Function Runtime Version: 4.27.5.21554

[2023-12-11T12:24:54.722Z] Customer packages not in sys path. This should never happen!
[2023-12-11T12:24:58.257Z] Worker process started and initialized.

Functions:

        queue_trigger: queueTrigger

For detailed output, run func with --verbose flag.
[2023-12-11T12:25:02.978Z] Host lock lease acquired by instance ID '00000000000000000000000xXXXXCC'.
[2023-12-11T12:25:31.670Z] Executing 'Functions.queue_trigger' (Reason='New queue message detected on 'queue123'.', Id=b7a1a111-XXXXXXXXXXXXX-769bbd7ace88)
[2023-12-11T12:25:31.677Z] Trigger Details: MessageId: 84b783dd-XXXXXXXXXXXXXXX04ae2656b, DequeueCount: 1, InsertedOn: 2023-12-11T12:25:12.000+00:00
[2023-12-11T12:25:31.817Z] Python Queue trigger processed a message: Hello world
[2023-12-11T12:25:31.891Z] Executed 'Functions.queue_trigger' (Succeeded, Id=b7a1a111-XXXXXXXXXXXXXXXX-769bbd7ace88, Duration=310ms)

Updated Solution:

To use an identity-based connection to a Storage Account which is not the AzureWebJobsStorage Account, assign the storage account as value to any new variable with pattern "<CONNECTION_NAME_PREFIX>__queueServiceUri": "https://<storage_account_name>.queue.core.windows.net"

local.settings.json:

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "QueueConn__queueServiceUri": "https://<your_storage_account_name>.queue.core.windows.net/"
  }
}

Code Snippet:

Change the connection value in the functionapp.py as connection =<CONNECTION_NAME_PREFIX>.

app = func.FunctionApp()


@app.queue_trigger(arg_name="azqueue", queue_name="<queuename>",
                               connection="QueueConn") 
def queue_trigger(azqueue: func.QueueMessage):
    logging.info('Python Queue trigger processed a message: %s',
                azqueue.get_body().decode('utf-8'))