🌐 Azure Event Grid¶

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

# Initialize client
credential = AzureKeyCredential(topic_key)
client = EventGridPublisherClient(topic_endpoint, credential)

# Create events
events = [
    EventGridEvent(
        event_type="user.created",
        data={
            "user_id": "user-123",
            "email": "user@example.com",
            "created_at": datetime.datetime.utcnow().isoformat()
        },
        subject="users/user-123",
        data_version="1.0"
    ),
    EventGridEvent(
        event_type="order.placed",
        data={
            "order_id": "order-456",
            "total_amount": 99.99,
            "items": 3
        },
        subject="orders/order-456",
        data_version="1.0"
    )
]

# Publish events
client.send(events)
print(f"Published {len(events)} events")

import logging
import json
import azure.functions as func

def main(event: func.EventGridEvent):
    """Process Event Grid event in Azure Function."""
    logging.info('Python EventGrid trigger processed an event')

    # Parse event
    event_data = event.get_json()

    logging.info(f'Event Type: {event.event_type}')
    logging.info(f'Event Subject: {event.subject}')
    logging.info(f'Event Data: {json.dumps(event_data, indent=2)}')

    # Process based on event type
    if event.event_type == "user.created":
        process_user_creation(event_data)
    elif event.event_type == "order.placed":
        process_order(event_data)
    else:
        logging.warning(f'Unknown event type: {event.event_type}')

def process_user_creation(data):
    """Handle user creation event."""
    user_id = data.get('user_id')
    email = data.get('email')

    # Send welcome email
    logging.info(f'Sending welcome email to {email}')

    # Create user profile
    logging.info(f'Creating profile for user {user_id}')

def process_order(data):
    """Handle order placement event."""
    order_id = data.get('order_id')
    total = data.get('total_amount')

    logging.info(f'Processing order {order_id} for ${total}')

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/api/events', methods=['POST', 'OPTIONS'])
def handle_event():
    """Webhook endpoint for Event Grid events."""
    if request.method == 'OPTIONS':
        # Handle validation handshake
        return '', 200

    if request.method == 'POST':
        events = request.get_json()

        # Handle validation event
        for event in events:
            if event['eventType'] == 'Microsoft.EventGrid.SubscriptionValidationEvent':
                validation_code = event['data']['validationCode']
                return jsonify({
                    'validationResponse': validation_code
                })

            # Process actual events
            process_event(event)

        return '', 200

def process_event(event):
    """Process individual event."""
    event_type = event['eventType']
    subject = event['subject']
    data = event['data']

    print(f"Event Type: {event_type}")
    print(f"Subject: {subject}")
    print(f"Data: {json.dumps(data, indent=2)}")

if __name__ == '__main__':
    app.run(port=5000)

# Method 1: Access Key
from azure.eventgrid import EventGridPublisherClient
from azure.core.credentials import AzureKeyCredential

credential = AzureKeyCredential(topic_key)
client = EventGridPublisherClient(endpoint, credential)

# Method 2: Managed Identity (Premium tier)
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = EventGridPublisherClient(endpoint, credential)

import hmac
import hashlib
import base64

def validate_event_grid_signature(request):
    """Validate Event Grid webhook signature."""
    # Get signature from header
    signature = request.headers.get('aeg-signature')

    # Get shared secret
    shared_secret = os.getenv('EVENTGRID_WEBHOOK_SECRET')

    # Compute expected signature
    payload = request.get_data()
    expected = base64.b64encode(
        hmac.new(
            shared_secret.encode('utf-8'),
            payload,
            hashlib.sha256
        ).digest()
    ).decode('utf-8')

    return hmac.compare_digest(signature, expected)

from azure.monitor.query import MetricsQueryClient
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta

credential = DefaultAzureCredential()
client = MetricsQueryClient(credential)

# Query Event Grid metrics
resource_id = "/subscriptions/{sub}/resourceGroups/rg-eventgrid/providers/Microsoft.EventGrid/topics/my-custom-topic"

response = client.query_resource(
    resource_id,
    metric_names=["PublishSuccessCount", "PublishFailCount", "DeliverySuccessCount"],
    timespan=timedelta(hours=1),
    granularity=timedelta(minutes=5)
)

for metric in response.metrics:
    print(f"{metric.name}: {metric.timeseries}")