Implement Webhooks API

[Wauplin]

Webhooks allows anyone to listen to users and repos on the Hub and get notified (on a webhook URL) when a repo gets updated (new commit, new discussion, new comment,...). They can be used to auto-convert models, build community bots, build CI/CD and much more!. For more info about webhooks, check out this guide.

Webhooks can be configured manually in the user settings. In addition, an API also exist to create webhooks programmatically. Integration in huggingface_hub shouldn't be too complex but I'd prefer to have some real use cases and feedback before moving forward on this (let's gauge the interest for it first).

I have added in first comment a first implementation that would help getting started with the API. Even though it's not shipped in huggingface_hub, it should be fully working. Please let me know you thoughts!

[Wauplin]

EDIT: added snippet for list_webhooks and get_webhook.

Usage:

# Watch all changes to the "HuggingFaceH4/zephyr-7b-beta" model
webhook = create_webhook(
    watched=[{"type": "model", "name": "HuggingFaceH4/zephyr-7b-beta"}],
    url="https://webhook.site/a2176e82-5720-43ee-9e06-f91cb4c91548",
    domains=["repo", "discussion"],
    secret="my-secret",
)["webhook"]

# Update webhook to subscribe to dataset changes as well + stop watching discussions
update_webhook(
    webhook_id=webhook["id"],
    # Subscribe to dataset changes as well
    watched=webhook["watched"] + [{"type": "dataset", "name": "HuggingFaceH4/ultrachat_200k"}],
    # Only watch repo changes (i.e. stop watching discussions)
    domains=["repo"],
    # Keep same url + secret
    url=webhook["url"],
    secret=webhook["secret"],
)

# Fetch existing webhooks
webhooks = list_webhooks()
webhook = get_webhook("63e38f5b7bb1e409d5bf1973")

# Temporarily disable webhook
disable_webhook(webhook["id"])

# Re-enable webhook
enable_webhook(webhook["id"])

# Delete webhook (non reversible)
delete_webhook(webhook["id"])

Implementation:

from typing import Dict, List, Literal, Optional, TypedDict

import requests

from huggingface_hub.utils import get_session, build_hf_headers, hf_raise_for_status


headers = build_hf_headers()
# or headers = build_hf_headers(token="hf_***")


class WatchedItem(TypedDict):
    # Examples:
    #     {"type": "user", "name": "julien-c"}
    #     {"type": "org", "name": "HuggingFaceH4"}
    #     {"type": "model", "name": "HuggingFaceH4/zephyr-7b-beta"}
    #     {"type": "dataset", "name": "HuggingFaceH4/ultrachat_200k"}
    #     {"type": "space", "name": "HuggingFaceH4/zephyr-chat"}
    type: Literal["model", "dataset", "space", "org", "user"]
    name: str


# Do you want to subscribe to repo updates (code changes), discussion updates (issues, PRs, comments), or both?
DOMAIN_T = Literal["repo", "discussion"]

def get_webhook(webhook_id: str) -> Dict:
    """Get a webhook by its id."""
    response = get_session().get(f"https://huggingface.co/api/settings/webhooks/{webhook_id}", headers=headers)
    hf_raise_for_status(response)
    return response.json()

def list_webhooks() -> List[Dict]:
    """List all configured webhooks."""
    response = get_session().get("https://huggingface.co/api/settings/webhooks", headers=headers)
    hf_raise_for_status(response)
    return response.json()

def create_webhook(watched: List[WatchedItem], url: str, domains: List[DOMAIN_T], secret: Optional[str]) -> Dict:
    """Create a new webhook.

    Args:
        watched (List[WatchedItem]):
            List of items to watch. It an be users, orgs, models, datasets or spaces.
            See `WatchedItem` for more details.
        url (str):
            URL to send the payload to.
        domains (List[Literal["repo", "discussion"]]):
            List of domains to watch. It can be "repo", "discussion" or both.
        secret (str, optional):
            Secret to use to sign the payload.

    Returns:
        dict: The created webhook.

    Example:
        ```python
        >>> payload = create_webhook(
        ...     watched=[{"type": "user", "name": "julien-c"}, {"type": "org", "name": "HuggingFaceH4"}],
        ...     url="https://webhook.site/a2176e82-5720-43ee-9e06-f91cb4c91548",
        ...     domains=["repo", "discussion"],
        ...     secret="my-secret",
        ... )
        {
            "webhook": {
                "id": "654bbbc16f2ec14d77f109cc",
                "watched": [{"type": "user", "name": "julien-c"}, {"type": "org", "name": "HuggingFaceH4"}],
                "url": "https://webhook.site/a2176e82-5720-43ee-9e06-f91cb4c91548",
                "secret": "my-secret",
                "domains": ["repo", "discussion"],
                "disabled": False,
            },
        }
        ```
    """
    response = get_session().post(
        "https://huggingface.co/api/settings/webhooks",
        json={"watched": watched, "url": url, "domains": domains, "secret": secret},
        headers=headers,
    )
    hf_raise_for_status(response)
    return response.json()


def update_webhook(
    webhook_id: str, watched: List[WatchedItem], url: str, domains: List[DOMAIN_T], secret: Optional[str]
) -> Dict:
    """Update an existing webhook.

    Exact same usage as `create_webhook` but you must know the `webhook_id`.
    All fields are updated.
    """
    response = get_session().post(
        f"https://huggingface.co/api/settings/webhooks/{webhook_id}",
        json={"watched": watched, "url": url, "domains": domains, "secret": secret},
        headers=headers,
    )
    hf_raise_for_status(response)
    return response.json()


def enable_webhook(webhook_id: str) -> Dict:
    """Enable a webhook (makes it "active")."""
    response = get_session().post(
        f"https://huggingface.co/api/settings/webhooks/{webhook_id}/enable",
        headers=headers,
    )
    hf_raise_for_status(response)
    return response.json()


def disable_webhook(webhook_id: str) -> Dict:
    """Disable a webhook (makes it "disabled")."""
    response = get_session().post(
        f"https://huggingface.co/api/settings/webhooks/{webhook_id}/disable",
        headers=headers,
    )
    hf_raise_for_status(response)
    return response.json()


def delete_webhook(webhook_id: str):
    """Delete a webhook."""
    response = get_session().delete(
        f"https://huggingface.co/api/settings/webhooks/{webhook_id}",
        headers=headers,
    )
    hf_raise_for_status(response)

[julien-c]

love the API!

[julien-c]

one question (i don't remember from the server-side implem) is there a way to list my webhooks or retrieve a webhook "listening" to a given watched repo or entity?

I'd expect this to be useful in flows of programmatic creation (to not re-create the same webhook many times)

[Wauplin]

one question (i don't remember from the server-side implem) is there a way to list my webhooks or retrieve a webhook "listening" to a given watched repo or entity?
I'd expect this to be useful in flows of programmatic creation (to not re-create the same webhook many times)

Not for now no (I think mostly because we did not have the need). I would also except something like:

# Get one from its id
def get_webhook(webhook_id:str) -> Dict:
    ...

# List all of my webhooks
def list_webhooks() -> List[Dict]:
    ...

I would expect both endpoints to be relatively easy to implement server-side. If we want to iterate on the list_webhooks afterwards (to search based on criteria), that's also possible but let's do a plain list first.

EDIT: APIs have been shipped server-side! We can now retrieve all configured webhooks or retrieve 1 by its id. I updated the code snippets in #1808 (comment).

[lunarflu]

Super cool @Wauplin ! 👀 🔥

[Wauplin]

Update: it is now possible to list all existing webhooks (or fetch 1 webhook from its id) 🚀. I have updated the code snippets above in #1808 (comment).

[lappemic]

Hey @Wauplin, following up on #1709, I would like to start working on this. I'm thinking of adding a the new module webhook_api with your code above within the huggingface_hub. Is this what you were thinking of as well?

[Wauplin]

Hi @lappemic, thanks for following-up on this issue! So actually it'd be good to add this stuff directly to the HfApi client that contains all Hub-related methods. This happens in the hf_api.py module. Methods would have to have self and token as input parameters as well + update their docstring to match the same standard in this class. PR https://github.com/huggingface/huggingface_hub/pull/1905/files is a good example PR that you can look at to get inspiration from. Please let me know if you have some further questions. Also don't hesitate to ping me on a draft PR if you want to start something and request feedback before continuing. Good luck!

[Wauplin]

Closed by #2209.