Python

@app.route(route="hello", methods=["GET", "POST"], auth_level=func.AuthLevel.ANONYMOUS)
def http_function(req: func.HttpRequest) -> func.HttpResponse:
    name = req.params.get('name') or req.get_body().decode()
    return func.HttpResponse(f"Hello, {name}!")

# Trigger (use EventGrid source)
@app.blob_trigger(arg_name="myblob", path="samples-workitems/{name}",
                  connection="AzureWebJobsStorage", source="EventGrid")
def blob_trigger(myblob: func.InputStream):
    logging.info(f"Blob: {myblob.name}, Size: {myblob.length}")

# Input
@app.blob_input(arg_name="inputblob", path="input/{name}", connection="AzureWebJobsStorage")

# Output
@app.blob_output(arg_name="outputblob", path="output/{name}", connection="AzureWebJobsStorage")

# Trigger
@app.queue_trigger(arg_name="msg", queue_name="myqueue-items",
                   connection="AzureWebJobsStorage")
def queue_trigger(msg: func.QueueMessage):
    logging.info(f"Queue message: {msg.get_body().decode()}")

# Output
@app.queue_output(arg_name="outputmsg", queue_name="outqueue",
                  connection="AzureWebJobsStorage")

@app.timer_trigger(schedule="0 */5 * * * *", arg_name="mytimer",
                   run_on_startup=False)
def timer_function(mytimer: func.TimerRequest):
    logging.info("Timer triggered")

# Trigger
@app.event_grid_trigger(arg_name="event")
def eventgrid_trigger(event: func.EventGridEvent):
    logging.info(f"Event: {event.subject} - {event.event_type}")

# Output
@app.event_grid_output(arg_name="outputEvent",
                       topic_endpoint_uri="MyEventGridTopicUriSetting",
                       topic_key_setting="MyEventGridTopicKeySetting")

# Trigger (Change Feed)
@app.cosmos_db_trigger_v3(arg_name="documents",
                          connection="CosmosDBConnection",
                          database_name="mydb",
                          container_name="mycontainer",
                          create_lease_container_if_not_exists=True)
def cosmosdb_trigger(documents: func.DocumentList):
    for doc in documents:
        logging.info(f"Changed doc: {doc['id']}")

# Input
@app.cosmos_db_input(arg_name="doc", connection="CosmosDBConnection",
                     database_name="mydb", container_name="mycontainer",
                     id="{id}", partition_key="{partitionKey}")

# Output
@app.cosmos_db_output(arg_name="newdoc", connection="CosmosDBConnection",
                      database_name="mydb", container_name="mycontainer")

# Queue Trigger
@app.service_bus_queue_trigger(arg_name="msg", queue_name="myqueue",
                               connection="ServiceBusConnection")
def sb_queue_trigger(msg: func.ServiceBusMessage):
    logging.info(f"Message: {msg.get_body().decode()}")

# Topic Trigger
@app.service_bus_topic_trigger(arg_name="msg", topic_name="mytopic",
                                subscription_name="mysubscription",
                                connection="ServiceBusConnection")
def sb_topic_trigger(msg: func.ServiceBusMessage):
    logging.info(f"Topic message: {msg.get_body().decode()}")

# Output
@app.service_bus_queue_output(arg_name="outmsg", queue_name="outqueue",
                              connection="ServiceBusConnection")

# Trigger
@app.event_hub_message_trigger(arg_name="event", event_hub_name="myeventhub",
                                connection="EventHubConnection",
                                cardinality="many")
def eventhub_trigger(event: func.EventHubEvent):
    logging.info(f"Event: {event.get_body().decode()}")

# Output
@app.event_hub_output(arg_name="outevent", event_hub_name="outeventhub",
                      connection="EventHubConnection")

# Input
@app.table_input(arg_name="tableEntity", table_name="mytable",
                 partition_key="{partitionKey}", row_key="{rowKey}",
                 connection="AzureWebJobsStorage")

# Output
@app.table_output(arg_name="tableOut", table_name="mytable",
                  connection="AzureWebJobsStorage")

# Trigger
@app.sql_trigger(arg_name="changes", table_name="dbo.MyTable",
                 connection_string_setting="SqlConnection")
def sql_trigger(changes: func.SqlRowList):
    for change in changes:
        logging.info(f"Change: {change}")

# Input
@app.sql_input(arg_name="items",
               command_text="SELECT * FROM dbo.MyTable WHERE Id = @Id",
               command_type="Text", parameters="@Id={id}",
               connection_string_setting="SqlConnection")

# Output
@app.sql_output(arg_name="newitem", command_text="dbo.MyTable",
                connection_string_setting="SqlConnection")

@app.generic_output_binding(arg_name="signalr", type="signalR",
                             hub_name="myhub",
                             connection_string_setting="AzureSignalRConnectionString")

@app.generic_output_binding(arg_name="mail", type="sendGrid",
                             api_key="SendGridApiKey",
                             from_address="noreply@example.com")

@app.route(route="process", methods=["POST"])
@app.cosmos_db_input(arg_name="doc", connection="CosmosDBConnection",
                     database_name="mydb", container_name="mycontainer",
                     id="{id}", partition_key="{pk}")
@app.queue_output(arg_name="outmsg", queue_name="outqueue",
                  connection="AzureWebJobsStorage")
def process_item(req: func.HttpRequest, doc: func.DocumentList,
                 outmsg: func.Out[str]) -> func.HttpResponse:
    outmsg.set(doc[0].to_json())
    return func.HttpResponse("Processed")