Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Azure Functions v2 Python Examples
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)
{
"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"
}
]
}
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
{
"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"
}
]
}
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)
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)
{
"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"
}
]
}
{
"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=...;"
}
}
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))
{
"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"
}
]
}
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}")
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "myTimer",
"type": "timerTrigger",
"direction": "in",
"schedule": "* */10 * * * *"
}
]
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.