How can I set Queue Storage message TTL in the context of an Azure Function output binding in Python?

Question

I have a python azure function with a queue output binding. I am successfully using this binding to queue messages from within the function. Is it possible to set the message TTL on the underlying queue or on the message itself? I don't need to set it on a per message basis, but will do it that way if that is the only option.

host.json

{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[1.*, 2.0.0)"
  }
}

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "type": "queue",
      "direction": "out",
      "name": "msg",
      "queueName": "predictions",
      "connection": "AzureWebJobsStorage"
    }
  ]
}

function code

import json
import logging
import azure.functions as func
from graphene import Schema
from .helpers import responses
from .schema.Query import Query

def main(req: func.HttpRequest, msg: func.Out[func.QueueMessage]) -> func.HttpResponse:
    logging.info('Executing GraphQL function.')

    try:
        query = req.get_body().decode()
    except ValueError:
        pass

    if query:
        schema = Schema(Query)
        results = schema.execute(query)
        response = responses.graphql(results)

        # Write response to azure queue storage
        message = responses.storage(query, response)
        if message:
            msg.set(message)

        return response
    else:
        return responses.bad_request(
            'Please pass a GraphQL query in the request body.')

George Chen George Chen · Accepted Answer · 2020-01-17T07:44:00

For now this only supports c# language, you could bind it to CloudQueue type. If it's other laguage you have to use the SDK method to implement. If you insist this feature, you could got to this github issue to comment you requirements.

And below is my test code to set TTL in a HTTP trigger function with azure-storage-queue 2.1.0.

import logging
import azure.functions as func
from azure.storage.queue import QueueService
import os

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    queue_service = QueueService(connection_string=os.environ['AzureWebJobsStorage'])

    message = req.params.get('message')
    if not message:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            message = req_body.get('message')

    if message:
        queue_service.put_message('myqueue',message,None,300,None)
        return func.HttpResponse(f" {message}!")
    else:
        return func.HttpResponse(
             "Please pass message in the request body",
             status_code=400
        )

How can I set Queue Storage message TTL in the context of an Azure Function output binding in Python?

1 Answers