Azure Eventgrid Py

import osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.eventgrid import EventGridPublisherClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>credential = DefaultAzureCredential(require_envvar=True)# Or use a specific credential directly in production:# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes# credential = ManagedIdentityCredential()
endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"
with EventGridPublisherClient(endpoint, credential) as client:    # Use client here (see following sections for operations)    ...

from azure.eventgrid import EventGridPublisherClient, CloudEventfrom azure.identity import DefaultAzureCredential
with EventGridPublisherClient(endpoint, DefaultAzureCredential()) as client:    # Single event    event = CloudEvent(        type="MyApp.Events.OrderCreated",        source="/myapp/orders",        data={"order_id": "12345", "amount": 99.99}    )    client.send(event)
    # Multiple events    events = [        CloudEvent(            type="MyApp.Events.OrderCreated",            source="/myapp/orders",            data={"order_id": f"order-{i}"}        )        for i in range(10)    ]    client.send(events)

from azure.eventgrid import EventGridEventfrom datetime import datetime, timezone
event = EventGridEvent(    subject="/myapp/orders/12345",    event_type="MyApp.Events.OrderCreated",    data={"order_id": "12345", "amount": 99.99},    data_version="1.0")
client.send(event)

event = CloudEvent(    type="MyApp.Events.ItemCreated",      # Required: event type    source="/myapp/items",                 # Required: event source    data={"key": "value"},                 # Event payload    subject="items/123",                   # Optional: subject/path    datacontenttype="application/json",   # Optional: content type    dataschema="https://schema.example",  # Optional: schema URL    time=datetime.now(timezone.utc),      # Optional: timestamp    extensions={"custom": "value"}         # Optional: custom attributes)

event = EventGridEvent(    subject="/myapp/items/123",            # Required: subject    event_type="MyApp.ItemCreated",        # Required: event type    data={"key": "value"},                 # Required: event payload    data_version="1.0",                    # Required: schema version    topic="/subscriptions/.../topics/...", # Optional: auto-set    event_time=datetime.now(timezone.utc)  # Optional: timestamp)

from azure.eventgrid.aio import EventGridPublisherClientfrom azure.identity.aio import DefaultAzureCredential
async def publish_events():    credential = DefaultAzureCredential()        async with EventGridPublisherClient(endpoint, credential) as client:        event = CloudEvent(            type="MyApp.Events.Test",            source="/myapp",            data={"message": "hello"}        )        await client.send(event)
import asyncioasyncio.run(publish_events())

from azure.eventgrid import EventGridPublisherClientfrom azure.identity import DefaultAzureCredential
# Namespace endpoint (different from custom topic)namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net"topic_name = "my-topic"
with EventGridPublisherClient(    endpoint=namespace_endpoint,    credential=DefaultAzureCredential()) as client:    client.send(        event,        namespace_topic=topic_name    )