Azure Event Grid SDK for Python。用于发布事件、处理CloudEvents和构建事件驱动架构。技能概述azure-eventgrid-py 技能提供了Azure Event Grid SDK for Python的完整使用指南。该技能帮助开发者构建事件驱动应用程序支持发布事件到Event Grid主题、使用CloudEvents和EventGridEvents模式、以及实现发布/订阅消息传递。下载地址https://github.com/sickn33/antigravity-awesome-skills/tree/main/skills/azure-eventgrid-py主要功能事件发布: 将事件发布到Event Grid主题和命名空间CloudEvents支持: 使用CNCF标准的CloudEvents 1.0模式EventGridEvents支持: 使用Azure原生事件模式命名空间主题: 支持Event Grid命名空间的拉取传递异步客户端: 支持异步操作提高性能触发条件在以下情况下应该调用此技能:需要构建事件驱动架构实现发布/订阅消息传递使用CloudEvents或EventGridEvents需要发布事件到Azure Event Grid实现事件路由和过滤安装方法pip install azure-eventgrid azure-identity环境变量EVENTGRID_TOPIC_ENDPOINThttps://topic-name.region.eventgrid.azure.net/api/events EVENTGRID_NAMESPACE_ENDPOINThttps://namespace.region.eventgrid.azure.net事件类型格式类用例Cloud Events 1.0CloudEvent标准、可互操作推荐Event Grid SchemaEventGridEventAzure原生格式使用示例示例1: 创建客户端from azure.identity import DefaultAzureCredentialfrom azure.eventgrid import EventGridPublisherClientcredential DefaultAzureCredential()endpoint https://topic-name.region.eventgrid.azure.net/api/eventsclient EventGridPublisherClient(endpoint, credential)示例2: 发布CloudEventsfrom azure.eventgrid import EventGridPublisherClient, CloudEventfrom azure.identity import DefaultAzureCredentialclient EventGridPublisherClient(endpoint, DefaultAzureCredential())# 单个事件event CloudEvent(typeMyApp.Events.OrderCreated,source/myapp/orders,data{order_id: 12345, amount: 99.99})client.send(event)# 多个事件events [CloudEvent(typeMyApp.Events.OrderCreated,source/myapp/orders,data{order_id: forder-{i}})for i in range(10)]client.send(events)示例3: 发布EventGridEventsfrom azure.eventgrid import EventGridEventfrom datetime import datetime, timezoneevent EventGridEvent(subject/myapp/orders/12345,event_typeMyApp.Events.OrderCreated,data{order_id: 12345, amount: 99.99},data_version1.0)client.send(event)示例4: CloudEvent属性event CloudEvent(typeMyApp.Events.ItemCreated, # 必需: 事件类型source/myapp/items, # 必需: 事件源data{key: value}, # 事件负载subjectitems/123, # 可选: 主题/路径datacontenttypeapplication/json, # 可选: 内容类型dataschemahttps://schema.example, # 可选: 模式URLtimedatetime.now(timezone.utc), # 可选: 时间戳extensions{custom: value} # 可选: 自定义属性)示例5: 异步客户端from azure.eventgrid.aio import EventGridPublisherClientfrom azure.identity.aio import DefaultAzureCredentialasync def publish_events():credential DefaultAzureCredential()async with EventGridPublisherClient(endpoint, credential) as client:event CloudEvent(typeMyApp.Events.Test,source/myapp,data{message: hello})await client.send(event)import asyncioasyncio.run(publish_events())示例6: 命名空间主题from azure.eventgrid.aio import EventGridPublisherClient# 命名空间端点不同于自定义主题namespace_endpoint https://namespace.region.eventgrid.azure.nettopic_name my-topicasync with EventGridPublisherClient(endpointnamespace_endpoint,credentialDefaultAzureCredential()) as client:await client.send(event,namespace_topictopic_name)最佳实践使用CloudEvents: 新应用程序使用CloudEvents行业标准批量事件: 发布多个事件时使用批量发送包含有意义的主题: 用于过滤和路由使用异步客户端: 用于高吞吐量场景处理重试: Event Grid内置重试机制设置适当的事件类型: 用于路由和过滤限制说明仅当任务明确符合上述范围时使用此技能不要将输出视为环境特定验证、测试或专家审查的替代品如果缺少所需的输入、权限、安全边界或成功标准请停止并寻求澄清