مكتبة عميل Azure Event Grid ل Python - الإصدار 4.17.0b1
Azure Event Grid هي خدمة توجيه أحداث ذكية مدارة بالكامل تسمح للاستهلاك الموحد للحدث باستخدام نموذج النشر والاشتراك.
التعليمات البرمجية | المصدرالحزمة (PyPI) | حزمة (Conda) | الوثائق | المرجعية لواجهة برمجة التطبيقاتوثائق | المنتجعينات | التغيير
إخلاء المسئولية
هذا إصدار بيتا من Azure EventGrid EventGridClient، والذي جنبا إلى جنب مع GA EventGridPublisherClient. EventGridClientيدعم publish_cloud_eventsالعمليات و acknowledge_cloud_eventsrelease_cloud_eventsreject_cloud_eventsrenew_cloud_event_locks. receive_cloud_events يرجى الرجوع إلى العينات لمزيد من المعلومات.
الشروع في العمل
المتطلبات الأساسية
- Python 3.7 أو أحدث مطلوب لاستخدام هذه الحزمة.
- يجب أن يكون لديك اشتراك Azure ومورد Event Grid Topic لاستخدام هذه الحزمة. اتبع هذا البرنامج التعليمي خطوة بخطوة لتسجيل موفر موارد Event Grid وإنشاء مواضيع Event Grid باستخدام مدخل Microsoft Azure. هناك برنامج تعليمي مشابه باستخدام Azure CLI.
تثبيت الحِزَمة
قم بتثبيت مكتبة عميل Azure Event Grid ل Python باستخدام pip:
pip install azure-eventgrid
- مطلوب موضوع أو مجال Event Grid موجود. يمكنك إنشاء المورد باستخدام مدخل Microsoft Azure أو Azure CLI
إذا كنت تستخدم Azure CLI، فاستبدل <resource-group-name> و <resource-name> بأسماء فريدة خاصة بك.
إنشاء Event Grid Topic
az eventgrid topic --create --location <location> --resource-group <resource-group-name> --name <resource-name>
إنشاء مجال شبكة الأحداث
az eventgrid domain --create --location <location> --resource-group <resource-group-name> --name <resource-name>
مصادقة العميل
للتفاعل مع خدمة Event Grid، ستحتاج إلى إنشاء مثيل لعميل. نقطة النهايةوبيانات الاعتماد ضرورية لإنشاء مثيل لكائن العميل.
استخدام Azure Active Directory (AAD)
توفر Azure Event Grid التكامل مع Azure Active Directory (Azure AD) للمصادقة المستندة إلى الهوية للطلبات. باستخدام Azure AD، يمكنك استخدام التحكم في الوصول استنادا إلى الدور (RBAC) لمنح حق الوصول إلى موارد Azure Event Grid للمستخدمين أو المجموعات أو التطبيقات.
لإرسال الأحداث إلى موضوع أو مجال باستخدام TokenCredential، يجب أن يكون للهوية المصادق عليها دور "مرسل بيانات EventGrid" المعين.
باستخدام الحزمة azure-identity ، يمكنك تخويل الطلبات بسلاسة في كل من بيئات التطوير والإنتاج. لمعرفة المزيد حول Azure Active Directory، راجع azure-identity README.
على سبيل المثال، يمكنك استخدام DefaultAzureCredential لإنشاء عميل يقوم بالمصادقة باستخدام Azure Active Directory:
from azure.identity import DefaultAzureCredential
from azure.eventgrid import EventGridPublisherClient, EventGridEvent
default_az_credential = DefaultAzureCredential()
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
client = EventGridPublisherClient(endpoint, default_az_credential)
البحث عن نقطة النهاية
يمكنك العثور على نقطة نهاية الموضوع داخل مورد Event Grid Topic على مدخل Microsoft Azure. سيبدو هذا كما يلي: "https://<event-grid-topic-name>.<topic-location>.eventgrid.azure.net/api/events"
إنشاء العميل باستخدام AzureKeyCredential
لاستخدام مفتاح Access كمعلمة credential ، قم بتمرير المفتاح كسلسلة إلى مثيل AzureKeyCredential.
ملاحظه: يمكن العثور على مفتاح الوصول في مدخل Microsoft Azure في قائمة "مفاتيح الوصول" لمورد Event Grid Topic. يمكن أيضا الحصول عليها عبر azure CLI أو المكتبة
azure-mgmt-eventgrid. يمكن العثور على دليل للحصول على مفاتيح الوصول هنا.
import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.credentials import AzureKeyCredential
topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
credential_key = AzureKeyCredential(topic_key)
client = EventGridPublisherClient(endpoint, credential_key)
ملاحظه: يمكن أيضا مصادقة العميل عبر توقيع SAS، باستخدام
AzureSasCredential. تتوفر عينة توضح ذلك هنا (async_version).
ملاحظه:
generate_sasيمكن استخدام الأسلوب لإنشاء توقيع وصول مشترك. يمكن رؤية عينة توضح هذا هنا.
المفاهيم الرئيسية
الموضوع
الموضوع هو قناة داخل خدمة EventGrid لإرسال الأحداث. يتم تحديد مخطط الحدث الذي يقبله الموضوع في وقت إنشاء الموضوع. إذا تم إرسال أحداث من نوع مخطط إلى موضوع يتطلب نوع مخطط مختلف، فسيتم رفع الأخطاء.
المجال
مجال الحدث هو أداة إدارة لعدد كبير من مواضيع Event Grid المتعلقة بنفس التطبيق. إنها تسمح لك بنشر الأحداث على آلاف الموضوعات. تمنحك المجالات أيضا التخويل والتحكم في المصادقة على كل موضوع. لمزيد من المعلومات، تفضل بزيارة نظرة عامة على مجال الحدث.
عند إنشاء مجال حدث، يتم توفير نقطة نهاية نشر لهذا المجال لك. تشبه هذه العملية إنشاء موضوع شبكة الأحداث. الفرق الوحيد هو أنه عند النشر إلى مجال، يجب تحديد الموضوع داخل المجال الذي تريد تسليم الحدث إليه.
مخططات الأحداث
الحدث هو أصغر كمية من المعلومات التي تصف بشكل كامل شيئا حدث في النظام. عند إنشاء موضوع أو مجال مخصص، يجب تحديد المخطط الذي سيتم استخدامه عند نشر الأحداث.
تدعم Event Grid مخططات متعددة لترميز الأحداث.
مخطط "شبكة الأحداث"
بينما يمكنك تكوين الموضوع لاستخدام مخطط مخصص، فمن الأكثر شيوعا استخدام مخطط Event Grid المحدد بالفعل. راجع المواصفات والمتطلبات هنا.
مخطط CloudEvents v1.0
خيار آخر هو استخدام مخطط CloudEvents v1.0. CloudEvents هو مشروع Cloud Native Computing Foundation الذي ينتج مواصفات لوصف بيانات الحدث بطريقة مشتركة. يمكن العثور على ملخص الخدمة ل CloudEvents هنا.
EventGridPublisherClient
EventGridPublisherClient يوفر عمليات لإرسال بيانات الحدث إلى اسم مضيف موضوع محدد أثناء تهيئة العميل.
بغض النظر عن المخطط الذي تم تكوين الموضوع أو المجال لاستخدامه، EventGridPublisherClient سيتم استخدامه لنشر الأحداث إليه. send استخدم أسلوب نشر الأحداث.
يسمح بإرسال التنسيقات التالية من الأحداث:
قائمة أو مثيل واحد من EventGridEvents مكتوبة بقوة.
تمثيل إملاء لعنصر EventGridEvent متسلسل.
قائمة أو مثيل واحد من CloudEvents مكتوبة بقوة.
تمثيل إملاء لكائن CloudEvent متسلسل.
تمثيل إملاء لأي مخطط مخصص.
يرجى إلقاء نظرة على العينات للحصول على أمثلة مفصلة.
ملاحظه: من المهم معرفة ما إذا كان موضوعك يدعم CloudEvents أو EventGridEvents قبل النشر. إذا قمت بإرسال إلى موضوع لا يدعم مخطط الحدث الذي ترسله، فسيطرح send() استثناء.
مواضيع النظام
يمثل موضوع النظام في Event Grid حدثا واحدا أو أكثر تم نشره بواسطة خدمات Azure مثل Azure Storage أو Azure Event Hubs. على سبيل المثال، قد يمثل موضوع النظام جميع أحداث الكائن الثنائي كبير الحجم أو أحداث إنشاء الكائن الثنائي كبير الحجم وحذفه فقط المنشورة لحساب تخزين معين.
تتوفر أسماء أنواع الأحداث المختلفة لأحداث النظام المنشورة على Azure Event Grid في azure.eventgrid.SystemEventNames.
للحصول على قائمة كاملة بمواضيع النظام التي يمكن التعرف عليها، تفضل بزيارة مواضيع النظام.
لمزيد من المعلومات حول المفاهيم الرئيسية على Event Grid، راجع المفاهيم في Azure Event Grid.
شبكة الأحداث على Kubernetes مع Azure Arc
شبكة الأحداث على Kubernetes مع Azure Arc هو عرض يسمح لك بتشغيل شبكة الأحداث على مجموعة Kubernetes الخاصة بك. يتم تمكين هذه الإمكانية من خلال استخدام Kubernetes الممكّنة في Azure Arc. من خلال Azure Arc تمكين Kubernetes، يتصل كتلة Kubernetes معتمدة إلى Azure. بمجرد الاتصال، يمكنك تثبيت شبكة الأحداث عليه. تعرف على المزيد حول هذا الأمر هنا.
دعم أحداث سحابة CNCF
بدءا من الإصدار 4.7.0، تدعم هذه الحزمة أيضا نشر حدث سحابة CNCF من https://pypi.org/project/cloudevents/. ستتمكن من تمرير كائن CloudEvent من هذه المكتبة إلى send واجهة برمجة التطبيقات.
from cloudevents.http import CloudEvent
event = CloudEvent(...)
client.send(event)
أمثلة
توفر الأقسام التالية العديد من القصاصات البرمجية التي تغطي بعض مهام شبكة الأحداث الأكثر شيوعا، بما في ذلك:
- إرسال حدث Event Grid
- إرسال حدث سحابة
- إرسال أحداث متعددة
- إرسال الأحداث كقواميس
- استهلاك حمولة من قائمة انتظار التخزين
- الاستهلاك من ServiceBus
إرسال حدث Event Grid
ينشر هذا المثال حدث Event Grid.
import os
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient, EventGridEvent
key = os.environ["EG_ACCESS_KEY"]
endpoint = os.environ["EG_TOPIC_HOSTNAME"]
event = EventGridEvent(
data={"team": "azure-sdk"},
subject="Door1",
event_type="Azure.Sdk.Demo",
data_version="2.0"
)
credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)
client.send(event)
إرسال حدث سحابي
ينشر هذا المثال حدث Cloud.
import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient
key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]
event = CloudEvent(
type="Azure.Sdk.Sample",
source="https://egsample.dev/sampleevent",
data={"team": "azure-sdk"}
)
credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)
client.send(event)
إرسال أحداث متعددة
من الممكن إرسال الأحداث كدفعة عند إرسال أحداث متعددة إلى موضوع أو مجال. يرسل هذا المثال قائمة CloudEvents باستخدام أسلوب الإرسال.
تحذير: عند إرسال قائمة بأحداث متعددة في وقت واحد، لن يؤدي التكرار وإرسال كل حدث إلى الأداء الأمثل. للحصول على أفضل أداء، يوصى بشدة بإرسال قائمة بالأحداث.
import os
from azure.core.credentials import AzureKeyCredential
from azure.core.messaging import CloudEvent
from azure.eventgrid import EventGridPublisherClient
key = os.environ["CLOUD_ACCESS_KEY"]
endpoint = os.environ["CLOUD_TOPIC_HOSTNAME"]
event0 = CloudEvent(
type="Azure.Sdk.Sample",
source="https://egsample.dev/sampleevent",
data={"team": "azure-sdk"}
)
event1 = CloudEvent(
type="Azure.Sdk.Sample",
source="https://egsample.dev/sampleevent",
data={"team2": "azure-eventgrid"}
)
events = [event0, event1]
credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)
client.send(events)
إرسال الأحداث كقواميس
يمكن أيضا استخدام تمثيل إملاء للنماذج المتسلسلة المعنية لنشر CloudEvent (أحداث) أو EventGridEvent (أحداث) بصرف النظر عن الكائنات مكتوبة بقوة.
استخدم تمثيلا يشبه الإملاء لإرساله إلى موضوع مع مخطط مخصص كما هو موضح أدناه.
import os
import uuid
import datetime as dt
from msrest.serialization import UTC
from azure.core.credentials import AzureKeyCredential
from azure.eventgrid import EventGridPublisherClient
key = os.environ["CUSTOM_SCHEMA_ACCESS_KEY"]
endpoint = os.environ["CUSTOM_SCHEMA_TOPIC_HOSTNAME"]
event = custom_schema_event = {
"customSubject": "sample",
"customEventType": "sample.event",
"customDataVersion": "2.0",
"customId": uuid.uuid4(),
"customEventTime": dt.datetime.now(UTC()).isoformat(),
"customData": "sample data"
}
credential = AzureKeyCredential(key)
client = EventGridPublisherClient(endpoint, credential)
client.send(event)
الاستهلاك من قائمة انتظار التخزين
يستهلك هذا المثال رسالة تم تلقيها من قائمة انتظار التخزين ويإلغاء تسلسلها إلى كائن CloudEvent.
from azure.core.messaging import CloudEvent
from azure.storage.queue import QueueServiceClient, BinaryBase64DecodePolicy
import os
import json
# all types of CloudEvents below produce same DeserializedEvent
connection_str = os.environ['STORAGE_QUEUE_CONN_STR']
queue_name = os.environ['STORAGE_QUEUE_NAME']
with QueueServiceClient.from_connection_string(connection_str) as qsc:
payload = qsc.get_queue_client(
queue=queue_name,
message_decode_policy=BinaryBase64DecodePolicy()
).peek_messages()
## deserialize payload into a list of typed Events
events = [CloudEvent.from_dict(json.loads(msg.content)) for msg in payload]
الاستهلاك من servicebus
يستهلك هذا المثال رسالة حمولة تم تلقيها من ServiceBus ويزيل تسلسلها إلى كائن EventGridEvent.
from azure.eventgrid import EventGridEvent
from azure.servicebus import ServiceBusClient
import os
import json
# all types of EventGridEvents below produce same DeserializedEvent
connection_str = os.environ['SERVICE_BUS_CONN_STR']
queue_name = os.environ['SERVICE_BUS_QUEUE_NAME']
with ServiceBusClient.from_connection_string(connection_str) as sb_client:
payload = sb_client.get_queue_receiver(queue_name).receive_messages()
## deserialize payload into a list of typed Events
events = [EventGridEvent.from_dict(json.loads(next(msg.body).decode('utf-8'))) for msg in payload]
التتبع الموزع باستخدام EventGrid
يمكنك استخدام OpenTelemetry ل Python كالمعتاد مع EventGrid لأنه متوافق مع تكامل التتبع azure-core.
فيما يلي مثال على استخدام OpenTelemetry لتتبع إرسال CloudEvent.
أولا، قم بتعيين OpenTelemetry كمكون إضافي ممكن للتتبع ل EventGrid.
from azure.core.settings import settings
from azure.core.tracing.ext.opentelemetry_span import OpenTelemetrySpan
settings.tracing_implementation = OpenTelemetrySpan
استخدام بيانات تتبع الاستخدام المفتوحة بانتظام من هنا. راجع OpenTelemetry للحصول على التفاصيل.
يستخدم هذا المثال مصدر وحدة تحكم بسيط لتصدير التتبعات. يمكن استخدام أي مصدر هنا بما في ذلك azure-monitor-opentelemetry-exporter، jaeger، zipkin وما إلى ذلك.
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor # this requires opentelemetry >= 1.0.0
# Simple console exporter
exporter = ConsoleSpanExporter()
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
trace.get_tracer_provider().add_span_processor(
SimpleSpanProcessor(exporter)
)
tracer بمجرد تعيين وexporter، يرجى اتباع المثال أدناه لبدء جمع التتبعات أثناء استخدام send الأسلوب من EventGridPublisherClient لإرسال كائن CloudEvent.
import os
from azure.eventgrid import EventGridPublisherClient
from azure.core.messaging import CloudEvent
from azure.core.credentials import AzureKeyCredential
hostname = os.environ['CLOUD_TOPIC_HOSTNAME']
key = AzureKeyCredential(os.environ['CLOUD_ACCESS_KEY'])
cloud_event = CloudEvent(
source = 'demo',
type = 'sdk.demo',
data = {'test': 'hello'},
)
with tracer.start_as_current_span(name="MyApplication"):
client = EventGridPublisherClient(hostname, key)
client.send(cloud_event)
استكشاف الأخطاء وإصلاحها
- تمكين
azure.eventgridالمسجل لجمع التتبعات من المكتبة.
عام
ستقوم مكتبة عميل Event Grid برفع الاستثناءات المحددة في Azure Core.
تسجيل الدخول
تستخدم هذه المكتبة مكتبة التسجيل القياسية للتسجيل. يتم تسجيل المعلومات الأساسية حول جلسات HTTP (عناوين URL والعناوين وما إلى ذلك) على مستوى INFO.
التكوين الاختياري
يمكن تمرير وسيطات الكلمات الأساسية الاختيارية على مستوى العميل وكل عملية. تصف الوثائق المرجعية azure-core التكوينات المتوفرة لإعادة المحاولة والتسجيل وبروتوكولات النقل والمزيد.
الخطوات التالية
يوفر القسم التالي العديد من القصاصات البرمجية التي توضح الأنماط الشائعة المستخدمة في واجهة برمجة تطبيقات Event Grid Python.
المزيد من نماذج التعليمات البرمجية
تعرض نماذج التعليمات البرمجية هذه عمليات سيناريو البطل الشائعة مع مكتبة عميل Azure Event Grid.
إنشاء توقيع الوصول المشترك: sample_generate_sas.py
مصادقة العميل: sample_authentication.py (async_version)
نشر الأحداث إلى موضوع باستخدام SAS: sample_publish_events_to_a_topic_using_sas_credential_async.py (async_version)
نشر أحداث شبكة الأحداث إلى موضوع: sample_publish_eg_events_to_a_topic.py (async_version)
نشر أحداث EventGrid إلى موضوع مجال: sample_publish_eg_events_to_a_domain_topic.py (async_version)
نشر حدث سحابي: sample_publish_events_using_cloud_events_1.0_schema.py (async_version)
نشر مخطط مخصص: sample_publish_custom_schema_to_a_topic.py (async_version)
تغطي العينات التالية تمثيلات النشر والاستهلاك dict ل EventGridEvents وCloudEvents.
نشر EventGridEvent كإملاء مثل التمثيل: sample_publish_eg_event_using_dict.py (async_version)
نشر CloudEvent كإملاء مثل التمثيل: sample_publish_cloud_event_using_dict.py (async_version)
استهلاك حمولة مخصصة من بيانات cloudevent الأولية: sample_consume_custom_payload.py
يمكن العثور على المزيد من العينات هنا.
- يمكن رؤية المزيد من العينات المتعلقة بسيناريو الإرسال هنا.
- لمشاهدة المزيد من العينات المتعلقة باستهلاك حمولة من خدمات مراسلة مختلفة ككائن مكتوب، يرجى زيارة استخدام العينات
وثائق إضافية
للحصول على وثائق أكثر شمولا حول Azure Event Grid، راجع وثائق Event Grid على docs.microsoft.com.
المساهمة
هذا المشروع يرحب بالمساهمات والاقتراحات. معظم المساهمات تتطلب منك الموافقة على اتفاقية ترخيص المساهم (CLA) التي تعلن أن لديك الحق في منحنا حق استخدام مساهمتك. للحصول على التفاصيل، تفضل بزيارة cla.microsoft.com.
عند إرسال طلب سحب، سيحدد روبوت CLA-bot تلقائيًا ما إذا كنت بحاجة إلى تقديم CLA وتزيين العلاقات العامة بشكل مناسب (على سبيل المثال، التسمية أو التعليق). ما عليك سوى اتباع التعليمات التي يقدمها الروبوت. ستحتاج فقط إلى القيام بذلك مرة واحدة عبر جميع عمليات إعادة الشراء باستخدام CLA الخاص بنا.
اعتمد هذا المشروع مدونة السلوك من المصادر المفتوحة من Microsoft. لمزيد من المعلومات، راجع الأسئلة المتداولة حول قواعد السلوك أو الاتصال opencode@microsoft.com بأي أسئلة أو تعليقات إضافية.
Azure SDK for Python
الملاحظات
قريبًا: خلال عام 2024، سنتخلص تدريجيًا من GitHub Issues بوصفها آلية إرسال ملاحظات للمحتوى ونستبدلها بنظام ملاحظات جديد. لمزيد من المعلومات، راجع https://aka.ms/ContentUserFeedback.
إرسال الملاحظات وعرضها المتعلقة بـ