Azure Automation Runbooks

# This is a Python runbook that runs in Azure Automation
import azure.mgmt.compute
import azure.mgmt.resource
from azure.identity import ManagedIdentityCredential

def main():
    credential = ManagedIdentityCredential()
    sub_id = "your-subscription-id"

    # List all VMs and their status
    compute_client = azure.mgmt.compute.ComputeManagementClient(credential, sub_id)

    for vm in compute_client.virtual_machines.list_all():
        instance_view = compute_client.virtual_machines.instance_view(
            vm.id.split("/")[4],  # resource group
            vm.name,
        )
        for status in instance_view.statuses:
            if status.code.startswith("PowerState/"):
                print(f"{vm.name}: {status.code}")

    # List resources in a resource group
    resource_client = azure.mgmt.resource.ResourceManagementClient(credential, sub_id)
    for resource in resource_client.resources.list_by_resource_group("my-rg"):
        print(f"{resource.type}: {resource.name}")

if __name__ == "__main__":
    main()

from azure.identity import DefaultAzureCredential
from azure.mgmt.automation import AutomationClient
from azure.mgmt.automation.models import (
    ScheduleCreateParameters,
    CreateOrUpdateJobScheduleParameters,
)
from datetime import datetime, timedelta

credential = DefaultAzureCredential()
client = AutomationClient(credential, "subscription-id")

# Create a daily schedule
schedule = client.schedule.create_or_update(
    "my-rg", "my-automation", "daily-2am",
    ScheduleCreate_parameters=ScheduleCreateParameters(
        name="daily-2am",
        description="Runs every day at 2 AM UTC",
        frequency="Day",
        interval=1,
        start_time=datetime.utcnow() + timedelta(hours=1),
        time_zone="UTC",
    ),
)

# Link schedule to a runbook
client.job_schedule.create_or_update(
    "my-rg", "my-automation", "daily-health-check-schedule",
    CreateOrUpdateJobScheduleParameters={
        "runbook": {"name": "health-check"},
        "schedule": {"name": "daily-2am"},
        "parameters": {"resource_group": "production"},
    },
)

from azure.identity import DefaultAzureCredential
from azure.mgmt.automation import AutomationClient
from azure.mgmt.automation.models import (
    RunbookCreateOrUpdateParameters,
    WebhookCreateOrUpdateParameters,
)

credential = DefaultAzureCredential()
client = AutomationClient(credential, "subscription-id")

# Create a runbook that targets a Hybrid Worker group
runbook = client.runbook.create_or_update(
    "my-rg", "my-automation", "onprem-backup",
    RunbookCreateOrUpdateParameters={
        "runbook_type": "Python3",
        "description": "Backup databases on on-premises servers",
        "log_progress": True,
        "log_verbose": False,
        "draft": {"content_type": "text/plain"},
    },
)

# Publish the runbook content
client.runbook.begin_publish(
    "my-rg", "my-automation", "onprem-backup"
).result()

# Start the runbook on a Hybrid Worker group
job = client.job.create(
    "my-rg", "my-automation", "onprem-backup-job",
    {
        "runbook": {"name": "onprem-backup"},
        "parameters": {"server": "db-server-01"},
        "run_on": "hybrid-worker-group-1",
    },
)

from azure.identity import DefaultAzureCredential
from azure.mgmt.automation import AutomationClient

credential = DefaultAzureCredential()
client = AutomationClient(credential, "subscription-id")

# Create a webhook
webhook = client.webhook.create_or_update(
    "my-rg", "my-automation", "deploy-webhook",
    {
        "name": "deploy-webhook",
        "description": "Triggers deployment runbook",
        "is_enabled": True,
        "runbook": {"name": "deploy-service"},
        "expiry_time": "2025-01-01T00:00:00Z",
    },
)

print(f"Webhook URL: {webhook.uri}")

# To trigger, POST to the webhook URL:
# curl -X POST "https://s12.azure-automation.net/webhooks?token=abc123" \
#      -H "Content-Type: application/json" \
#      -d '{"service": "web-01", "replicas": 3}'

import logging
import sys
from datetime import datetime

# Configure logging for Azure Automation
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger(__name__)

def main():
    try:
        logger.info("Starting runbook execution")
        result = perform_operation()
        logger.info(f"Runbook completed: {result}")
    except Exception as e:
        logger.error(f"Runbook failed: {e}")
        # Write output for the runbook history
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)

def perform_operation() -> dict:
    """The actual runbook logic."""
    # Your automation code here
    return {"status": "success", "timestamp": datetime.utcnow().isoformat()}

if __name__ == "__main__":
    main()