azure.eventgrid.aio package¶
-
class
azure.eventgrid.aio.
EventGridPublisherClient
(endpoint: str, credential: Union[AsyncTokenCredential, azure.core.credentials.AzureKeyCredential, azure.core.credentials.AzureSasCredential], **kwargs: Any)[source]¶ Asynchronous EventGridPublisherClient publishes events to an EventGrid topic or domain. It can be used to publish either an EventGridEvent, a CloudEvent or a Custom Schema.
- Parameters
endpoint (str) – The topic endpoint to send the events to.
credential (AzureKeyCredential or AzureSasCredential or AsyncTokenCredential) – The credential object used for authentication which implements SAS key authentication or SAS token authentication or an AsyncTokenCredential.
- Return type
Example:
import os from azure.eventgrid.aio import EventGridPublisherClient from azure.core.credentials import AzureKeyCredential topic_key = os.environ["EVENTGRID_TOPIC_KEY"] endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"] credential = AzureKeyCredential(topic_key) client = EventGridPublisherClient(endpoint, credential)
import os from azure.eventgrid.aio import EventGridPublisherClient from azure.core.credentials import AzureSasCredential signature = os.environ["EVENTGRID_SAS"] endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"] credential = AzureSasCredential(signature) client = EventGridPublisherClient(endpoint, credential)
-
async
close
() → None[source]¶ Close the
EventGridPublisherClient
session.
-
async
send
(events: Union[azure.core.messaging.CloudEvent, azure.eventgrid._models.EventGridEvent, Dict, List[azure.core.messaging.CloudEvent], List[azure.eventgrid._models.EventGridEvent], List[Dict]], *, channel_name: Optional[str] = None, **kwargs: Any) → None[source]¶ Sends events to a topic or a domain specified during the client initialization.
A single instance or a list of dictionaries, CloudEvents or EventGridEvents are accepted.
Example:
import os import asyncio from azure.eventgrid import EventGridEvent from azure.eventgrid.aio import EventGridPublisherClient from azure.core.credentials import AzureKeyCredential topic_key = os.environ["EVENTGRID_TOPIC_KEY"] endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"] async def publish(): credential = AzureKeyCredential(topic_key) client = EventGridPublisherClient(endpoint, credential) await client.send([ EventGridEvent( event_type="Contoso.Items.ItemReceived", data={ "itemSku": "Contoso Item SKU #1" }, subject="Door1", data_version="2.0" ) ])
import os import asyncio from azure.core.messaging import CloudEvent from azure.eventgrid.aio import EventGridPublisherClient from azure.core.credentials import AzureKeyCredential topic_key = os.environ["EVENTGRID_CLOUD_EVENT_TOPIC_KEY"] endpoint = os.environ["EVENTGRID_CLOUD_EVENT_TOPIC_ENDPOINT"] async def publish(): credential = AzureKeyCredential(topic_key) client = EventGridPublisherClient(endpoint, credential) await client.send([ CloudEvent( type="Contoso.Items.ItemReceived", source="/contoso/items", data={ "itemSku": "Contoso Item SKU #1" }, subject="Door1" ) ])
Dict representation of respective serialized models is accepted as CloudEvent(s) or EventGridEvent(s) apart from the strongly typed objects.
Example:
event0 = { "eventType": "Contoso.Items.ItemReceived", "data": { "itemSku": "Contoso Item SKU #1" }, "subject": "Door1", "dataVersion": "2.0", "id": "randomuuid11", "eventTime": datetime.utcnow() } event1 = { "eventType": "Contoso.Items.ItemReceived", "data": { "itemSku": "Contoso Item SKU #2" }, "subject": "Door1", "dataVersion": "2.0", "id": "randomuuid12", "eventTime": datetime.utcnow() } async with client: await client.send([event0, event1])
async with client: await client.send([ { "type": "Contoso.Items.ItemReceived", "source": "/contoso/items", "data": { "itemSku": "Contoso Item SKU #1" }, "subject": "Door1", "specversion": "1.0", "id": "randomclouduuid11" } ])
When publishing a Custom Schema Event(s), dict-like representation is accepted. Either a single dictionary or a list of dictionaries can be passed.
Example:
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" } async with client: # publish list of events await client.send(custom_schema_event)
WARNING: When sending a list of multiple events at one time, iterating over and sending each event will not result in optimal performance. For best performance, it is highly recommended to send a list of events.
- Parameters
events (CloudEvent or EventGridEvent or dict or List[CloudEvent] or List[EventGridEvent] or List[dict]) – A single instance or a list of dictionaries/CloudEvent/EventGridEvent to be sent.
- Keyword Arguments
content_type (str) – The type of content to be used to send the events. Has default value “application/json; charset=utf-8” for EventGridEvents, with “cloudevents-batch+json” for CloudEvents
channel_name (str) – Optional. Used to specify the name of event channel when publishing to partner namespaces with partner topic. For more details, visit https://docs.microsoft.com/azure/event-grid/partner-events-overview
- Return type