Create a webhook automation

import wandb

# Initialize the API
api = wandb.Api()

# List all webhook integrations for your team
webhook_integrations = list(api.webhook_integrations(entity="your-team"))

if not webhook_integrations:
    print("❌ No webhook integrations found!")
    print("Please configure a webhook in your team settings:")
    print(f"https://wandb.ai/{your-team}/settings/webhooks")
else:
    print(f"✅ Found {len(webhook_integrations)} webhook integration(s):")
    for webhook in webhook_integrations:
        print(f"   - Name: {webhook.name}")
        print(f"     URL: {webhook.url_endpoint}")
        print(f"     ID: {webhook.id}")
        if hasattr(webhook, 'has_access_token') and webhook.has_access_token:
            print(f"     Auth: Has access token")

def get_webhook_integration(entity, name_pattern=None, url_pattern=None):
    """Get a webhook integration by name or URL pattern"""
    webhooks = list(api.webhook_integrations(entity=entity))
    
    if not webhooks:
        raise ValueError(f"No webhook integrations found for {entity}. "
                        "Please configure one in team settings.")
    
    # Filter by name pattern
    if name_pattern:
        matching = [w for w in webhooks if name_pattern in w.name]
        if matching:
            return matching[0]
    
    # Filter by URL pattern
    if url_pattern:
        matching = [w for w in webhooks if url_pattern in w.url_endpoint]
        if matching:
            return matching[0]
    
    # Return first webhook if no pattern or no matches
    return webhooks[0]

def list_webhook_requirements(webhook):
    """Display information about webhook configuration"""
    print(f"Webhook: {webhook.name}")
    print(f"  Endpoint: {webhook.url_endpoint}")
    
    if hasattr(webhook, 'has_access_token') and webhook.has_access_token:
        print("  ✓ Has access token configured")
        print("    Use ${ACCESS_TOKEN} in payload to access it")
    
    if hasattr(webhook, 'secrets') and webhook.secrets:
        print(f"  ✓ Has {len(webhook.secrets)} secret(s) configured")
        for secret in webhook.secrets:
            print(f"    Use ${{{secret}}} in payload to access secret '{secret}'")

# Example usage
try:
    # Get webhook for GitHub integration
    github_webhook = get_webhook_integration("my-team", url_pattern="github.com")
    list_webhook_requirements(github_webhook)
except ValueError as e:
    print(e)

# Check if webhook supports authentication
webhook = get_webhook_integration("my-team", "deployment-webhook")

# Build payload based on webhook configuration
payload = {
    "event": "model_ready",
    "model": "${artifact_version_string}",
}

# Add authentication if configured
if hasattr(webhook, 'has_access_token') and webhook.has_access_token:
    payload["auth"] = "${ACCESS_TOKEN}"  # W&B will substitute the actual token

# Create automation with appropriate payload
action = SendWebhook.from_integration(webhook, payload=payload)

# List all automations for a project or registry
automations = project.automations()  # or registry.automations()

# Filter automations by type
webhook_automations = [
    a for a in automations 
    if a.action.action_type == "GENERIC_WEBHOOK"
]

# Update webhook payload
for automation in webhook_automations:
    if automation.name == "my-webhook-automation":
        # Modify the automation configuration
        automation.action.request_payload = updated_payload
        automation.save()

# Delete an automation
automation.delete()

# Assuming you have a secret named "DEPLOY_TOKEN" configured
payload = {
    "auth_token": "${DEPLOY_TOKEN}",  # Secret accessed with $ prefix
    "deployment": {
        "model": "${artifact_version_string}",
        "environment": "production"
    }
}

# Get multiple webhook integrations
staging_webhook = next(
    w for w in api.webhook_integrations(entity="my-team")
    if "staging" in w.name
)
production_webhook = next(
    w for w in api.webhook_integrations(entity="my-team")
    if "production" in w.name
)

# Create different automations for different conditions
# Staging automation for experiments
staging_event = OnRunMetric(
    scope=project,
    filter=RunEvent.metric("val_accuracy") > 0.90
)
staging_automation = api.create_automation(
    staging_event >> SendWebhook.from_integration(staging_webhook),
    name="staging-deployment"
)

# Production automation for higher threshold
prod_event = OnRunMetric(
    scope=project,
    filter=RunEvent.metric("val_accuracy") > 0.95
)
prod_automation = api.create_automation(
    prod_event >> SendWebhook.from_integration(production_webhook),
    name="production-deployment"
)

try:
    # Verify webhook integration exists
    webhook_integrations = list(api.webhook_integrations(entity="my-team"))
    if not webhook_integrations:
        raise ValueError("No webhook integrations found. Please configure webhooks first.")
    
    # Create automation with error handling
    automation = api.create_automation(
        event >> action,
        name="validated-webhook-automation",
        description="Webhook automation with validation"
    )
    print(f"Successfully created: {automation.name}")
    
except Exception as e:
    print(f"Failed to create automation: {e}")