Webhooks

class WebhookClient:
    def get(self) -> dict | None:
        """Get webhook information."""
    
    def update(self, **kwargs) -> dict:
        """Update webhook configuration.
        
        Args:
            event_types (list[str], optional): List of event types to listen for
            request_url (str, optional): URL to send webhook requests to
            payload_template (str, optional): Custom payload template
            headers_template (str, optional): Custom headers template
            condition (dict, optional): Conditions for webhook firing
            **kwargs: Additional webhook parameters
        """
    
    def delete(self) -> None:
        """Delete webhook."""
    
    def test(self) -> dict | None:
        """Test webhook by sending dummy payload."""
    
    def dispatches(self) -> WebhookDispatchCollectionClient:
        """Get webhook dispatches collection."""

class WebhookClientAsync:
    """Async version of WebhookClient with identical methods."""

class WebhookCollectionClient:
    def list(
        self, 
        *, 
        limit: int | None = None, 
        offset: int | None = None, 
        desc: bool | None = None
    ) -> ListPage[dict]:
        """List webhooks.
        
        Args:
            limit: Maximum number of webhooks
            offset: Pagination offset
            desc: Sort in descending order
        """
    
    def create(self, **kwargs) -> dict:
        """Create new webhook.
        
        Args:
            event_types (list[str]): Event types to listen for
            request_url (str): Target URL for webhook requests
            payload_template (str, optional): Custom payload template
            headers_template (str, optional): Custom headers template
            condition (dict, optional): Webhook conditions
            **kwargs: Additional webhook configuration
        """

class WebhookCollectionClientAsync:
    """Async version of WebhookCollectionClient with identical methods."""

class WebhookDispatchClient:
    def get(self) -> dict | None:
        """Get webhook dispatch information."""

class WebhookDispatchClientAsync:
    """Async version of WebhookDispatchClient with identical methods."""

class WebhookDispatchCollectionClient:
    def list(
        self, 
        *, 
        limit: int | None = None, 
        offset: int | None = None, 
        desc: bool | None = None
    ) -> ListPage[dict]:
        """List webhook dispatches.
        
        Args:
            limit: Maximum number of dispatches
            offset: Pagination offset
            desc: Sort in descending order
        """

class WebhookDispatchCollectionClientAsync:
    """Async version of WebhookDispatchCollectionClient with identical methods."""

from apify_client import ApifyClient

client = ApifyClient('your-api-token')

# Create webhook for Actor run events
webhook = client.webhooks().create(
    event_types=['ACTOR.RUN.SUCCEEDED', 'ACTOR.RUN.FAILED'],
    request_url='https://myapp.com/webhook/apify',
    payload_template='{"runId": "{{runId}}", "status": "{{status}}", "actorId": "{{actorId}}"}',
    headers_template='{"Authorization": "Bearer {{myToken}}", "Content-Type": "application/json"}'
)

# Test webhook
webhook_client = client.webhook(webhook['id'])
test_result = webhook_client.test()
print(f"Webhook test result: {test_result}")

# Monitor webhook dispatches
dispatches = webhook_client.dispatches().list(limit=10)
for dispatch in dispatches.items:
    print(f"Dispatch {dispatch['id']}: {dispatch['status']}")