Webhook Management

def begin_create(resource_group_name: str, registry_name: str, webhook_name: str, webhook_create_parameters: WebhookCreateParameters, **kwargs) -> LROPoller[Webhook]:
    """
    Create a webhook for a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    - webhook_create_parameters: WebhookCreateParameters - Webhook configuration
    
    Returns:
    LROPoller[Webhook] - Long-running operation poller for the webhook
    """

def begin_delete(resource_group_name: str, registry_name: str, webhook_name: str, **kwargs) -> LROPoller[None]:
    """
    Delete a webhook from a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook to delete
    
    Returns:
    LROPoller[None] - Long-running operation poller
    """

def begin_update(resource_group_name: str, registry_name: str, webhook_name: str, webhook_update_parameters: WebhookUpdateParameters, **kwargs) -> LROPoller[Webhook]:
    """
    Update a webhook for a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook to update
    - webhook_update_parameters: WebhookUpdateParameters - Update parameters
    
    Returns:
    LROPoller[Webhook] - Long-running operation poller for the updated webhook
    """

def get(resource_group_name: str, registry_name: str, webhook_name: str, **kwargs) -> Webhook:
    """
    Get properties of a webhook.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    
    Returns:
    Webhook - Webhook resource with complete configuration
    """

def list(resource_group_name: str, registry_name: str, **kwargs) -> ItemPaged[Webhook]:
    """
    List all webhooks for a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    
    Returns:
    ItemPaged[Webhook] - Paginated list of all webhooks for the registry
    """

def get_callback_config(resource_group_name: str, registry_name: str, webhook_name: str, **kwargs) -> CallbackConfig:
    """
    Get callback configuration for a webhook.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    
    Returns:
    CallbackConfig - Webhook callback configuration details
    """

def update_callback_config(resource_group_name: str, registry_name: str, webhook_name: str, callback_config: CallbackConfig, **kwargs) -> CallbackConfig:
    """
    Update callback configuration for a webhook.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    - callback_config: CallbackConfig - New callback configuration
    
    Returns:
    CallbackConfig - Updated callback configuration
    """

def list_events(resource_group_name: str, registry_name: str, webhook_name: str, **kwargs) -> ItemPaged[Event]:
    """
    List recent events for a webhook.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    
    Returns:
    ItemPaged[Event] - Paginated list of recent webhook events
    """

def ping(resource_group_name: str, registry_name: str, webhook_name: str, **kwargs) -> EventInfo:
    """
    Trigger a ping event for a webhook.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - webhook_name: str - Name of the webhook
    
    Returns:
    EventInfo - Information about the ping event
    """

class Webhook:
    """
    An object that represents a webhook for a container registry.
    
    Attributes:
    - id: str - Resource ID
    - name: str - Resource name
    - type: str - Resource type
    - location: str - Resource location
    - tags: Dict[str, str] - Resource tags
    - status: WebhookStatus - Webhook status (enabled/disabled)
    - scope: str - Repository scope for the webhook
    - actions: List[WebhookAction] - Event actions that trigger the webhook
    - provisioning_state: ProvisioningState - Current provisioning state
    """

class WebhookCreateParameters:
    """
    Parameters for creating a webhook.
    
    Attributes:
    - tags: Dict[str, str] - Resource tags
    - location: str - Webhook location
    - service_uri: str - Service URI for the webhook endpoint
    - custom_headers: Dict[str, str] - Custom headers to include in webhook calls
    - status: WebhookStatus - Webhook status (enabled/disabled)
    - scope: str - Repository scope (e.g., 'myrepo:*' or 'myrepo:tag')
    - actions: List[WebhookAction] - Actions that trigger the webhook
    """

class WebhookUpdateParameters:
    """
    Parameters for updating a webhook.
    
    Attributes:
    - tags: Dict[str, str] - Resource tags
    - service_uri: str - Service URI for the webhook endpoint
    - custom_headers: Dict[str, str] - Custom headers
    - status: WebhookStatus - Webhook status
    - scope: str - Repository scope
    - actions: List[WebhookAction] - Actions that trigger the webhook
    """

class CallbackConfig:
    """
    Configuration for webhook callback.
    
    Attributes:
    - service_uri: str - Service endpoint URI
    - custom_headers: Dict[str, str] - Custom headers sent with webhook calls
    """

class Event:
    """
    Event for a webhook.
    
    Attributes:
    - id: str - Event ID
    - event_request_message: EventRequestMessage - Request message details
    - event_response_message: EventResponseMessage - Response message details
    """

class EventRequestMessage:
    """
    Event request message for webhook.
    
    Attributes:
    - content: EventContent - Event content with registry and action details
    - headers: Dict[str, str] - Request headers
    - method: str - HTTP method
    - request_uri: str - Request URI
    - version: str - API version
    """

class EventContent:
    """
    Content of a webhook event.
    
    Attributes:
    - id: str - Event ID
    - timestamp: datetime - Event timestamp
    - action: str - Action that triggered the event
    - target: Dict - Target object (repository, tag, digest details)
    - request: Dict - Request details (id, addr, host, method, useragent)
    - actor: Actor - Actor who performed the action
    - source: Dict - Source registry information
    """

class WebhookAction(str, Enum):
    """Actions that can trigger webhooks."""
    PUSH = "push"
    DELETE = "delete"
    QUARANTINE = "quarantine"
    CHART_PUSH = "chart_push"
    CHART_DELETE = "chart_delete"

class WebhookStatus(str, Enum):
    """Webhook status options."""
    ENABLED = "enabled"
    DISABLED = "disabled"

from azure.mgmt.containerregistry import ContainerRegistryManagementClient
from azure.mgmt.containerregistry.models import (
    WebhookCreateParameters, WebhookAction, WebhookStatus
)
from azure.identity import DefaultAzureCredential

client = ContainerRegistryManagementClient(
    DefaultAzureCredential(), 
    "subscription-id"
)

# Create a webhook that triggers on image pushes and deletions
webhook_params = WebhookCreateParameters(
    location="East US",
    service_uri="https://mycicd.example.com/webhook/registry",
    status=WebhookStatus.ENABLED,
    actions=[WebhookAction.PUSH, WebhookAction.DELETE],
    scope="myapp/*",  # Only for repositories starting with 'myapp'
    custom_headers={
        "Authorization": "Bearer your-token-here",
        "X-Source": "azure-container-registry"
    },
    tags={
        "purpose": "cicd-integration",
        "environment": "production"
    }
)

# Create the webhook
creation_poller = client.webhooks.begin_create(
    "my-resource-group",
    "my-registry",
    "cicd-webhook",
    webhook_params
)

webhook = creation_poller.result()
print(f"Created webhook: {webhook.name}")
print(f"Status: {webhook.status}")
print(f"Actions: {webhook.actions}")

# Create a webhook for security scanning on image pushes
security_webhook_params = WebhookCreateParameters(
    location="East US",
    service_uri="https://security-scanner.example.com/scan",
    status=WebhookStatus.ENABLED,
    actions=[WebhookAction.PUSH],  # Only on pushes
    scope="*",  # All repositories
    custom_headers={
        "X-API-Key": "your-scanner-api-key",
        "X-Scan-Type": "vulnerability",
        "Content-Type": "application/json"
    },
    tags={
        "purpose": "security-scanning",
        "priority": "high"
    }
)

security_webhook = client.webhooks.begin_create(
    "my-resource-group",
    "my-registry", 
    "security-scanner",
    security_webhook_params
).result()

print(f"Security webhook created: {security_webhook.name}")

# List recent webhook events
events = client.webhooks.list_events(
    "my-resource-group",
    "my-registry",
    "cicd-webhook"
)

print("Recent Webhook Events:")
print("-" * 40)
for event in events:
    if event.event_request_message and event.event_request_message.content:
        content = event.event_request_message.content
        print(f"Event ID: {event.id}")
        print(f"Action: {content.action}")
        print(f"Timestamp: {content.timestamp}")
        print(f"Repository: {content.target.get('repository', 'N/A')}")
        print(f"Tag: {content.target.get('tag', 'N/A')}")
        
        if event.event_response_message:
            print(f"Response Status: {event.event_response_message.status_code}")
        print("-" * 40)

# Ping webhook to test connectivity
ping_result = client.webhooks.ping(
    "my-resource-group",
    "my-registry",
    "cicd-webhook"
)

print(f"Ping Event ID: {ping_result.id}")

# Check if ping was successful by looking at recent events
import time
time.sleep(2)  # Wait for event to be processed

recent_events = list(client.webhooks.list_events(
    "my-resource-group",
    "my-registry", 
    "cicd-webhook"
))

ping_event = next((e for e in recent_events if e.id == ping_result.id), None)
if ping_event and ping_event.event_response_message:
    status_code = ping_event.event_response_message.status_code
    print(f"Webhook ping response: {status_code}")
    if status_code == "200":
        print("Webhook is responding correctly")
    else:
        print(f"Webhook returned error: {status_code}")

from azure.mgmt.containerregistry.models import WebhookUpdateParameters

# Update webhook to include chart operations and change scope
update_params = WebhookUpdateParameters(
    actions=[
        WebhookAction.PUSH, 
        WebhookAction.DELETE, 
        WebhookAction.CHART_PUSH,
        WebhookAction.CHART_DELETE
    ],
    scope="production/*",  # Only production repositories
    custom_headers={
        "Authorization": "Bearer updated-token",
        "X-Source": "azure-container-registry",
        "X-Environment": "production"
    },
    tags={
        "purpose": "cicd-integration",
        "environment": "production",
        "updated": "2024-01-15"
    }
)

updated_webhook = client.webhooks.begin_update(
    "my-resource-group",
    "my-registry",
    "cicd-webhook",
    update_params
).result()

print(f"Updated webhook actions: {updated_webhook.actions}")
print(f"Updated webhook scope: {updated_webhook.scope}")

# Create webhooks for different purposes
webhook_configs = [
    {
        "name": "dev-cicd",
        "scope": "dev/*",
        "uri": "https://dev-cicd.example.com/webhook",
        "actions": [WebhookAction.PUSH]
    },
    {
        "name": "prod-cicd", 
        "scope": "prod/*",
        "uri": "https://prod-cicd.example.com/webhook",
        "actions": [WebhookAction.PUSH, WebhookAction.DELETE]
    },
    {
        "name": "security-scan",
        "scope": "*",
        "uri": "https://security.example.com/scan",
        "actions": [WebhookAction.PUSH]
    }
]

created_webhooks = []
for config in webhook_configs:
    webhook_params = WebhookCreateParameters(
        location="East US",
        service_uri=config["uri"],
        status=WebhookStatus.ENABLED,
        actions=config["actions"],
        scope=config["scope"],
        tags={"purpose": config["name"]}
    )
    
    webhook = client.webhooks.begin_create(
        "my-resource-group",
        "my-registry",
        config["name"],
        webhook_params
    ).result()
    
    created_webhooks.append(webhook)
    print(f"Created {config['name']}: {webhook.scope} -> {config['uri']}")

# List all webhooks
all_webhooks = client.webhooks.list("my-resource-group", "my-registry")
print(f"\nTotal webhooks: {len(list(all_webhooks))}")