How to publish events in azure event grid to a topic in python?

Question

Unable to import the eventgrid modules in my VS Code, I have added all the modules in requirement.txt and pip installed from my cmd.

Also, below is the python function which I am looking and trying with:

def publish_event(): 
    # authenticate client 
    credential = AzureKeyCredential(key) 
    client = EventGridPublisherClient(endpoint, credential) 
 
    custom_schema_event = { 
        "customSubject": "sample", 
        "customEventType": "sample.event", 
        "customDataVersion": "2.0", 
        "customId": uuid.uuid4(), 
        "customEventTime": dt.datetime.now(UTC()).isoformat(), 
        "customData": "sample data" 
    } 
 
    # publish events 
    for _  in range(3): 
 
        event_list = []     # list of events to publish 
        # create events and append to list 
        for j in range(randint(1, 3)): 
            event_list.append(custom_schema_event) 
 
        # publish list of events 
        client.send(event_list) 
        print("Batch of size {} published".format(len(event_list))) 
        time.sleep(randint(1, 5)) 
 
 
if name == '__main__': 
    publish_event()

I am not sure whether this is the correct method to achieve this, looking for better ways to solve the module issue and to publish events.

Help would be appreciated !

score 2 · Accepted Answer · answered Nov 26 '21 at 12:07

From your question we can see there are two points which needs to be answered:

Python packages installation issue
Eventgrid topic issue

For installing Azure Python packages with pip install, we need to enable virtual environment in our terminal, commands are as follows:
```
   python -m venv .venv
  .venv\Scripts\activate
```
After enabling the venv, add all eventgrid, storage and other azure related packages in requirements.txt, then run below command in cmd:

pip install -r requirements.txt

Now, looking at your code, function looks fine as it is sending the events to a general topic. In a similar way we have a sample which gives us some insight about publishing the eventgrid events to a custom topic. This was explained in Azure-SDK-Python-GitHub repo. Python code as below:

  import os
  from random import randint, sample
  import time

  from azure.core.credentials import AzureKeyCredential
  from azure.eventgrid import EventGridPublisherClient, EventGridEvent

  key = os.environ["EG_ACCESS_KEY"]
  endpoint = os.environ["EG_TOPIC_HOSTNAME"]

  #authenticate client
  credential = AzureKeyCredential(key)
  client = EventGridPublisherClient(endpoint, credential)
  services = ["EventGrid", "ServiceBus", "EventHubs", "Storage"]    #possible values for data field

  def publish_event():
      #publish events
      for _ in range(3):

          event_list = []     #list of events to publish
          #create events and append to list
          for j in range(randint(1, 3)):
              sample_members = sample(services, k=randint(1, 4))      #select random subset of team members
              event = EventGridEvent(
                      subject="Door1",
                      data={"team": sample_members},
                      event_type="Azure.Sdk.Demo",
                      data_version="2.0"
                      )
              event_list.append(event)

          #publish list of events
          client.send(event_list)
          print("Batch of size {} published".format(len(event_list)))
          time.sleep(randint(1, 5))

  if __name__ == '__main__':
      publish_event()

How to publish events in azure event grid to a topic in python?

1 Answers1