How to locally develop EventHub Triggered Functions in Python (programming model v2)?

I would like to learn to develop Azure Functions locally using Visual Studio Code. While there are numerous examples and demos available for using an HTTP trigger, I'm struggling to find much information on creating a minimal working example for event-triggered functions. I've even come across some disagreements regarding whether it's possible to develop locally without connecting to an actual Event Hub Service.

I have a few questions for the community:

1. Is it feasible to develop event-triggered functions locally (and with a reasonable effort)?
2. If anyone has successfully done this, could you please provide an example? I've gone through several posts, but I'm having trouble putting everything together. I saw a mention of "Thunder Client", but I'm unfamiliar with it. Could someone explain if it's an option and how it works?
3. What should the host.json and local.settings.json files look like?

I would like to start with the sample test code provided by Microsoft. Here is the code:

import azure.functions as func
import logging

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger")
@app.event_hub_message_trigger(arg_name="hub", 
                               event_hub_name="<EVENT_HUB_NAME>",
                               connection="<CONNECTION_SETTING>") 

def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

I appreciate any guidance or assistance you can provide. Thank you!

1. Is it feasible to develop event-triggered functions locally (and with a reasonable effort)?

Yes, Its feasible and easy to run Azure Event Hub trigger function locally and also if you deploy the same function in Function app Portal.

2. If anyone has successfully done this, could you please provide an example? I’ve gone through several posts, but I’m having trouble putting everything together. I saw a mention of “Thunder Client”, but I’m unfamiliar with it. Could someone explain if it’s an option and how it works?

Azure Event Hub trigger as the name suggests will trigger after your Event hub receives an Event.

My Azure Event Hub trigger function ran successfully, Locally refer below:-

Steps to create Event Hub function and send events in Event Hub, You can also refer this MS Document:-

Create a new Function trigger with Azure Event Hub Trigger like below in your VS Code, Make sure you have Azure Function extension installed:-

I opened one Folder in my Vs Code and creates a new Function trigger like below:-

Now create a local.settings.json like below:-

Select Create new local app setting and Select your Eventhubnamespace like below:-

Select your EventhubNamespace like below:-

Select eventhub like below:-

Select consumer group as Default:-

Event Hub trigger got created successfully:-

Now, Open a new Folder in your VS Code in another terminal and add the below code to send the Events.

I have referred the below code from this MS Document but I have added a below code to send the data as a JSON instead of string as Azure event Hub trigger only triggers when JSON based data is passed in Event hub:-

eventsender.py

import  json

  

from  azure.eventhub  import  EventData

from  azure.eventhub.aio  import  EventHubProducerClient

  

EVENT_HUB_CONNECTION_STR = "eventhubconnectionstring"

EVENT_HUB_NAME = "siliconeventhub"<eventhub name not event hub
namespace>

  

async  def  run():

# Create a producer client to send messages to the event hub.

# Specify a connection string to your event hubs namespace and

# the event hub name.

producer = EventHubProducerClient.from_connection_string(

conn_str=EVENT_HUB_CONNECTION_STR, eventhub_name=EVENT_HUB_NAME

)

async  with  producer:

# Create a batch.

event_data_batch = await  producer.create_batch()

  

# Add events to the batch.

event_data_batch.add(EventData(json.dumps({"message": "First
event"})))

event_data_batch.add(EventData(json.dumps({"message": "Second
event"})))

event_data_batch.add(EventData(json.dumps({"message": "Third
event"})))

  

# Send the batch of events to the event hub.

await  producer.send_batch(event_data_batch)

  

asyncio.run(run()) 

Run your Event Hub trigger like below:-

Click fn + f5 or Click on run > Debug > It will prompt you to connect to a storage account like below:-

Select the storage account like below:-

Event Hub trigger Function ran successfully like below:-

Run the eventsender.py code in another tab like below:-

My Function triggered successfully like below:-

My event hub namespace reference:-

Connection String reference MS Document :-

My eventhub with Default consumer group:-

UPDATED Programming Model V2:-

Yes, I am referring to Azure Event hub Trigger. I created one Azure Event Hub trigger with Python programming model v2 like below:-

I tried setting UseDevelopmentStorage:true value in local.settings.json but it asked me to connect to Storage account when I ran it as I did not have Azure storage emulator and azurite extension installed, In order to run your function code with UseDevelopmentStorage:true install Azurite extension and install Azure Storage emulator from this Link and start it in your VS code and your function will run without a need to connect to storage account, even if it asks you to connect to storage account you can click on Debug anyway.

Refer below:-

local.settings.json:-

{

"IsEncrypted": false,

"Values": {

"FUNCTIONS_WORKER_RUNTIME": "python",

"AzureWebJobsStorage": "UseDevelopmentStorage=true",

"AzureWebJobsFeatureFlags": "EnableWorkerIndexing",

"connection":"Endpoint=sb://siliconeventhub.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=zSxxxxNOrjxxxxxxxx/bnxxxxI4d1gQ="

}

}

Install Azurite Extension in your VS Code for UseDevelopmentStorage=true to work like below:-

My host.json:-

{

"version": "2.0",

"logging": {

"applicationInsights": {

"samplingSettings": {

"isEnabled": true,

"excludedTypes": "Request"

}

}

},

"extensionBundle": {

"id": "Microsoft.Azure.Functions.ExtensionBundle",

"version": "[3.15.0, 4.0.0)"

}

}

My functionapp.py code referred from this MS Document:-

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger1")
@app.event_hub_message_trigger(arg_name="myhub", 
                               event_hub_name="siliconeventhub",
                               connection="connection") 
def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

I have added connection="connection" which retrieves value from local.settings.json.

Now, I run the Function and Event Hub triggered successfully with local Azure storage emulator and settings "AzureWebJobsStorage": "UseDevelopmentStorage=true" Refer below:-

Click Fn + F5 or Run > Start Debugging to run your V2 Function or click on Debug anyway when this pop up come as we are already using Azurite and Azure Storage emulator

Note- In order to resolve above warning, If you have your Azure storage emulator started with the steps below it won't give you debug anyway warning like above.

Now, I started my Azure Storage emulator by searching it in start menu like below:-

Azure storage emulator started like below:-

Azure Event hub triggered successfully like below with v2 programming model.