🌐 System Topics¶

# Python Azure Function to process blob events
import logging
import json
import azure.functions as func
from azure.storage.blob import BlobServiceClient

def main(event: func.EventGridEvent):
    """Process blob creation events."""
    result = json.dumps({
        'id': event.id,
        'data': event.get_json(),
        'topic': event.topic,
        'subject': event.subject,
        'event_type': event.event_type,
    })

    logging.info(f'Python EventGrid trigger processed an event: {result}')

    # Extract blob details
    data = event.get_json()
    blob_url = data['url']
    blob_size = data['contentLength']

    logging.info(f'New blob created: {blob_url} ({blob_size} bytes)')

    # Process the blob
    if blob_url.endswith('.csv'):
        process_csv_file(blob_url)
    elif blob_url.endswith('.json'):
        process_json_file(blob_url)

def process_csv_file(blob_url):
    """Process CSV file from blob storage."""
    blob_service = BlobServiceClient.from_connection_string(
        os.environ["STORAGE_CONNECTION_STRING"]
    )

    # Download and process blob
    blob_client = blob_service.get_blob_client_from_url(blob_url)
    blob_data = blob_client.download_blob()
    content = blob_data.readall()

    # Process CSV content
    logging.info(f"Processing CSV: {len(content)} bytes")

# Azure Function to handle IoT device events
import azure.functions as func
import logging
from azure.cosmos import CosmosClient

def main(event: func.EventGridEvent):
    """Handle IoT Hub device events."""
    event_type = event.event_type
    device_data = event.get_json()

    device_id = device_data.get('deviceId')
    logging.info(f"Device event: {event_type} for device {device_id}")

    if event_type == 'Microsoft.Devices.DeviceConnected':
        handle_device_connected(device_id, device_data)

    elif event_type == 'Microsoft.Devices.DeviceDisconnected':
        handle_device_disconnected(device_id, device_data)

    elif event_type == 'Microsoft.Devices.DeviceCreated':
        handle_device_created(device_id, device_data)

    elif event_type == 'Microsoft.Devices.DeviceDeleted':
        handle_device_deleted(device_id, device_data)

def handle_device_connected(device_id, data):
    """Update device status when connected."""
    logging.info(f"Device {device_id} connected")

    # Update device registry in Cosmos DB
    cosmos_client = CosmosClient.from_connection_string(
        os.environ["COSMOS_CONNECTION_STRING"]
    )
    database = cosmos_client.get_database_client("iot")
    container = database.get_container_client("devices")

    device_record = {
        "id": device_id,
        "status": "online",
        "lastConnected": data.get('eventTime'),
        "hubName": data.get('hubName')
    }

    container.upsert_item(device_record)
    logging.info(f"Updated device {device_id} status to online")

def handle_device_disconnected(device_id, data):
    """Update device status when disconnected."""
    logging.info(f"Device {device_id} disconnected")

    # Update device registry
    cosmos_client = CosmosClient.from_connection_string(
        os.environ["COSMOS_CONNECTION_STRING"]
    )
    database = cosmos_client.get_database_client("iot")
    container = database.get_container_client("devices")

    device_record = {
        "id": device_id,
        "status": "offline",
        "lastDisconnected": data.get('eventTime')
    }

    container.upsert_item(device_record)

# Azure Function for Key Vault secret expiry
import azure.functions as func
import logging
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

def main(event: func.EventGridEvent):
    """Handle Key Vault secret events."""
    event_type = event.event_type
    data = event.get_json()

    secret_name = data.get('ObjectName')
    vault_name = data.get('VaultName')

    logging.info(f"Key Vault event: {event_type} for secret {secret_name}")

    if event_type == 'Microsoft.KeyVault.SecretNearExpiry':
        rotate_secret(vault_name, secret_name)

    elif event_type == 'Microsoft.KeyVault.SecretExpired':
        handle_expired_secret(vault_name, secret_name)

def rotate_secret(vault_name, secret_name):
    """Rotate secret before expiration."""
    logging.warning(f"Secret {secret_name} expiring soon, rotating...")

    vault_url = f"https://{vault_name}.vault.azure.net"
    credential = DefaultAzureCredential()
    client = SecretClient(vault_url=vault_url, credential=credential)

    # Generate new secret value (example: API key rotation)
    new_secret_value = generate_new_api_key()

    # Update the secret
    client.set_secret(secret_name, new_secret_value)

    logging.info(f"Rotated secret {secret_name}")

    # Notify administrators
    send_notification(f"Secret {secret_name} has been rotated")

def handle_expired_secret(vault_name, secret_name):
    """Handle expired secret alert."""
    logging.error(f"Secret {secret_name} has EXPIRED!")

    # Send critical alert
    send_critical_alert(f"CRITICAL: Secret {secret_name} in vault {vault_name} has expired")

# Python webhook handler for ACR events
from fastapi import FastAPI, Request
import logging

app = FastAPI()

@app.post("/api/trigger-pipeline")
async def handle_acr_event(request: Request):
    """Handle ACR image push events."""
    event_data = await request.json()

    for event in event_data:
        event_type = event.get('eventType')

        if event_type == 'Microsoft.ContainerRegistry.ImagePushed':
            data = event.get('data')
            image = f"{data['repository']}:{data['tag']}"

            logging.info(f"New image pushed: {image}")

            # Trigger deployment pipeline
            await trigger_deployment_pipeline(
                image=image,
                registry=data['registry'],
                repository=data['repository'],
                tag=data['tag']
            )

    return {"status": "processed"}

async def trigger_deployment_pipeline(image, registry, repository, tag):
    """Trigger Azure DevOps pipeline."""
    # Call Azure DevOps API to trigger pipeline
    logging.info(f"Triggering deployment for {image}")