1# Python — Azure Functions v2 Triggers & Bindings
2 
3> **Model**: Python v2 programming model. **NO** `function.json` files.
4> Entry point: `function_app.py`
5> Import: `import azure.functions as func`
6 
7## Lambda Migration Rules
8 
9> Shared rules (bindings over SDKs, latest runtime, identity-first auth) → [global-rules.md](../global-rules.md)
10 
11Python-specific: all functions use the v2 decorator model shown throughout this file. No additional migration rules beyond global.
12 
13## HTTP Trigger
14 
15```python
16@app.route(route="hello", methods=["GET", "POST"], auth_level=func.AuthLevel.ANONYMOUS)
17def http_function(req: func.HttpRequest) -> func.HttpResponse:
18    name = req.params.get('name') or req.get_body().decode()
19    return func.HttpResponse(f"Hello, {name}!")
20```
21 
22## Blob Storage
23 
24```python
25# Trigger (use EventGrid source)
26@app.blob_trigger(arg_name="myblob", path="samples-workitems/{name}",
27                  connection="AzureWebJobsStorage", source="EventGrid")
28def blob_trigger(myblob: func.InputStream):
29    logging.info(f"Blob: {myblob.name}, Size: {myblob.length}")
30 
31# Input
32@app.blob_input(arg_name="inputblob", path="input/{name}", connection="AzureWebJobsStorage")
33 
34# Output
35@app.blob_output(arg_name="outputblob", path="output/{name}", connection="AzureWebJobsStorage")
36```
37 
38## Queue Storage
39 
40```python
41# Trigger
42@app.queue_trigger(arg_name="msg", queue_name="myqueue-items",
43                   connection="AzureWebJobsStorage")
44def queue_trigger(msg: func.QueueMessage):
45    logging.info(f"Queue message: {msg.get_body().decode()}")
46 
47# Output
48@app.queue_output(arg_name="outputmsg", queue_name="outqueue",
49                  connection="AzureWebJobsStorage")
50```
51 
52## Timer
53 
54```python
55@app.timer_trigger(schedule="0 */5 * * * *", arg_name="mytimer",
56                   run_on_startup=False)
57def timer_function(mytimer: func.TimerRequest):
58    logging.info("Timer triggered")
59```
60 
61## Event Grid
62 
63```python
64# Trigger
65@app.event_grid_trigger(arg_name="event")
66def eventgrid_trigger(event: func.EventGridEvent):
67    logging.info(f"Event: {event.subject} - {event.event_type}")
68 
69# Output
70@app.event_grid_output(arg_name="outputEvent",
71                       topic_endpoint_uri="MyEventGridTopicUriSetting",
72                       topic_key_setting="MyEventGridTopicKeySetting")
73```
74 
75## Cosmos DB
76 
77```python
78# Trigger (Change Feed)
79@app.cosmos_db_trigger_v3(arg_name="documents",
80                          connection="CosmosDBConnection",
81                          database_name="mydb",
82                          container_name="mycontainer",
83                          create_lease_container_if_not_exists=True)
84def cosmosdb_trigger(documents: func.DocumentList):
85    for doc in documents:
86        logging.info(f"Changed doc: {doc['id']}")
87 
88# Input
89@app.cosmos_db_input(arg_name="doc", connection="CosmosDBConnection",
90                     database_name="mydb", container_name="mycontainer",
91                     id="{id}", partition_key="{partitionKey}")
92 
93# Output
94@app.cosmos_db_output(arg_name="newdoc", connection="CosmosDBConnection",
95                      database_name="mydb", container_name="mycontainer")
96```
97 
98## Service Bus
99 
100```python
101# Queue Trigger
102@app.service_bus_queue_trigger(arg_name="msg", queue_name="myqueue",
103                               connection="ServiceBusConnection")
104def sb_queue_trigger(msg: func.ServiceBusMessage):
105    logging.info(f"Message: {msg.get_body().decode()}")
106 
107# Topic Trigger
108@app.service_bus_topic_trigger(arg_name="msg", topic_name="mytopic",
109                                subscription_name="mysubscription",
110                                connection="ServiceBusConnection")
111def sb_topic_trigger(msg: func.ServiceBusMessage):
112    logging.info(f"Topic message: {msg.get_body().decode()}")
113 
114# Output
115@app.service_bus_queue_output(arg_name="outmsg", queue_name="outqueue",
116                              connection="ServiceBusConnection")
117```
118 
119## Event Hubs
120 
121```python
122# Trigger
123@app.event_hub_message_trigger(arg_name="event", event_hub_name="myeventhub",
124                                connection="EventHubConnection",
125                                cardinality="many")
126def eventhub_trigger(event: func.EventHubEvent):
127    logging.info(f"Event: {event.get_body().decode()}")
128 
129# Output
130@app.event_hub_output(arg_name="outevent", event_hub_name="outeventhub",
131                      connection="EventHubConnection")
132```
133 
134## Table Storage
135 
136```python
137# Input
138@app.table_input(arg_name="tableEntity", table_name="mytable",
139                 partition_key="{partitionKey}", row_key="{rowKey}",
140                 connection="AzureWebJobsStorage")
141 
142# Output
143@app.table_output(arg_name="tableOut", table_name="mytable",
144                  connection="AzureWebJobsStorage")
145```
146 
147## SQL
148 
149```python
150# Trigger
151@app.sql_trigger(arg_name="changes", table_name="dbo.MyTable",
152                 connection_string_setting="SqlConnection")
153def sql_trigger(changes: func.SqlRowList):
154    for change in changes:
155        logging.info(f"Change: {change}")
156 
157# Input
158@app.sql_input(arg_name="items",
159               command_text="SELECT * FROM dbo.MyTable WHERE Id = @Id",
160               command_type="Text", parameters="@Id={id}",
161               connection_string_setting="SqlConnection")
162 
163# Output
164@app.sql_output(arg_name="newitem", command_text="dbo.MyTable",
165                connection_string_setting="SqlConnection")
166```
167 
168## SignalR
169 
170```python
171@app.generic_output_binding(arg_name="signalr", type="signalR",
172                             hub_name="myhub",
173                             connection_string_setting="AzureSignalRConnectionString")
174```
175 
176## SendGrid
177 
178```python
179@app.generic_output_binding(arg_name="mail", type="sendGrid",
180                             api_key="SendGridApiKey",
181                             from_address="[email protected]")
182```
183 
184## Combining Decorators
185 
186```python
187@app.route(route="process", methods=["POST"])
188@app.cosmos_db_input(arg_name="doc", connection="CosmosDBConnection",
189                     database_name="mydb", container_name="mycontainer",
190                     id="{id}", partition_key="{pk}")
191@app.queue_output(arg_name="outmsg", queue_name="outqueue",
192                  connection="AzureWebJobsStorage")
193def process_item(req: func.HttpRequest, doc: func.DocumentList,
194                 outmsg: func.Out[str]) -> func.HttpResponse:
195    outmsg.set(doc[0].to_json())
196    return func.HttpResponse("Processed")
197```
198 
199> Full reference: [Azure Functions Python developer guide](https://learn.microsoft.com/en-us/azure/azure-functions/functions-reference-python)
200