Azure Functions v2 Python Examples

blobexample-__init__.py

import logging
import azure.functions as func

def main(inputBlob: func.InputStream, outputBlob: func.Out[str]):

    logging.info(f"Blob trigger executed!")
    logging.info(f"Blob Name: {inputBlob.name} ({inputBlob.length}) bytes")
    logging.info(f"Full Blob URI: {inputBlob.uri}")

    output = "Hello World!"
    outputBlob.set(output)

blobexample-function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "inputBlob",
      "type": "blobTrigger",
      "direction": "in",
      "path": "myblobcontainer/input/{name}",
      "connection": "MyStorageConnectionAppSetting"
    },
    {
      "name": "outputBlob",
      "type": "blob",
      "direction": "out",
      "path": "myblobcontainer/output/{name}",
      "connection": "MyStorageConnectionAppSetting"
    }
  ]
}

cosmosdbexample-__init__.py

import logging
import azure.functions as func

def main(documents: func.DocumentList) -> func.Document:

    logging.info(f"CosmosDB trigger executed!")

    for doc in documents:
        logging.info(f"Document: {doc.to_json()}")

    returnDoc = func.Document.from_dict({"text": "Hello World", "foo": "bar"})
    return returnDoc

cosmosdbexample-function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "documents",
      "type": "cosmosDBTrigger",
      "direction": "in",
      "databaseName": "cosmostest",
      "collectionName": "mycollection",
      "connectionStringSetting": "MyCosmosDBConnectionAppSetting",
      "leaseCollectionName": "leases",
      "createLeaseCollectionIfNotExists": true
    },
    {
      "name": "$return",
      "type": "cosmosDB",
      "direction": "out",
      "databaseName": "cosmostest",
      "collectionName": "mycollection2",
      "connectionStringSetting": "MyCosmosDBConnectionAppSetting"
    }
  ]
}

cosmosdbmultioutputexample-__init__.py

import logging
import azure.functions as func

def main(documents: func.DocumentList) -> func.DocumentList:

    logging.info(f"CosmosDB trigger executed!")

    for doc in documents:
        logging.info(f"Document: {doc.to_json()}")

    returnDocs = []
    for x in range(0, 3):
        newDoc = func.Document.from_dict({"text": str(x), "foo": "bar"})
        returnDocs.append(newDoc)

    return func.DocumentList(returnDocs)

httpexample-__init__.py

import logging
import azure.functions as func

def main(request: func.HttpRequest) -> func.HttpResponse:

    logging.info(f"HTTP trigger executed!")

    logging.info(f"Headers: {request.headers}")
    logging.info(f"Params: {request.params}")
    logging.info(f"Route Params: {request.route_params}")
    logging.info(f"Body: {request.get_body()}")

    try:
        logging.info(f"Body JSON: {request.get_json()}")
    except ValueError:
        pass

    return func.HttpResponse(f"Hello World!", status_code=200)

httpexample-function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "request",
      "type": "httpTrigger",
      "direction": "in",
      "authLevel": "function",
      "methods": [
        "get",
        "post"
      ],
      "route": "HttpExample/{user:alpha?}/{id:int?}"
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    }
  ]
}

localdotsettings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "DefaultEndpointsProtocol=https;AccountName=xxx;AccountKey=...",
    "MyStorageConnectionAppSetting": "DefaultEndpointsProtocol=https;AccountName=xxx;AccountKey=...",
    "MyCosmosDBConnectionAppSetting": "AccountEndpoint=https://xxx.documents.azure.com:443/;AccountKey=...;"
  }
}

queueexample-__init__.py

import logging
import json
import azure.functions as func

def main(inMessage: func.QueueMessage, outMessage: func.Out[func.QueueMessage]):

    logging.info(f"Queue trigger executed!")
    logging.info(f"Message Id: {inMessage.id}")
    logging.info(f"Message Body: {inMessage.get_body()}")
    logging.info(f"Message Body as JSON: {inMessage.get_json()}")
    logging.info(f"Dequeue Count: {inMessage.dequeue_count}")
    logging.info(f"Insertion Time: {inMessage.insertion_time}")
    logging.info(f"Expiration Time: {inMessage.expiration_time}")
    logging.info(f"Time Next Visible: {inMessage.time_next_visible}")

    output = {'text': 'abc'}
    outMessage.set(json.dumps(output))

queueexample-function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "inMessage",
      "type": "queueTrigger",
      "direction": "in",
      "queueName": "my-input-queue",
      "connection": "MyStorageConnectionAppSetting"
    },
    {
      "name": "outMessage",
      "type": "queue",
      "direction": "out",
      "queueName": "my-output-queue",
      "connection": "MyStorageConnectionAppSetting"
    }
  ]
}

timerexample-__init__.py

import datetime
import logging
import azure.functions as func

def main(myTimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info(f"Python timer trigger function ran at {utc_timestamp}")

timerexample-function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "* */10 * * * *"
    }
  ]
}