Hub Python 库文档

Webhooks 服务器

Hugging Face's logo
加入 Hugging Face 社区

并获得增强的文档体验

开始使用

Webhooks 服务器

Webhooks 是 MLOps 相关功能的基础。它们允许您监听特定仓库或您感兴趣的所有用户/组织仓库的新变化。要了解有关 Huggingface Hub 上 Webhooks 的更多信息,您可以阅读 Webhooks 指南

请查看此指南,了解如何设置 Webhooks 服务器并将其部署为 Space 的分步教程。

这是一个实验性功能。这意味着我们仍在努力改进 API。未来可能会在不事先通知的情况下引入破坏性更改。请确保在您的要求中固定 `huggingface_hub` 的版本。当您使用实验性功能时,会触发警告。您可以通过将 `HF_HUB_DISABLE_EXPERIMENTAL_WARNING=1` 设置为环境变量来禁用它。

服务器

该服务器是一个 Gradio 应用。它有一个用于为您或您的用户显示指令的 UI,以及一个用于侦听 webhook 的 API。实现 webhook 端点就像装饰函数一样简单。然后,您可以通过将 Webhook 重定向到您的机器(使用 Gradio 隧道)来调试它,然后再将其部署到 Space。

WebhooksServer

huggingface_hub.WebhooksServer

< >

( *args **kwargs )

参数

  • ui (gradio.Blocks, optional) — 用于作为 Space 登录页面的 Gradio UI 实例。如果为 None,则会创建一个显示有关已配置 webhook 的说明的 UI。
  • webhook_secret (str, optional) — 用于验证传入 webhook 请求的密钥。您可以将此值设置为您想要的任何密钥,只要您也在 webhooks 设置面板中配置它即可。您也可以将此值设置为 WEBHOOK_SECRET 环境变量。如果未提供密钥,则 webhook 端点将以不安全的方式打开。

WebhooksServer() 类允许您创建一个 Gradio 应用程序实例,该应用程序可以接收 Huggingface webhook。这些 webhook 可以使用 `add_webhook()` 装饰器进行注册。Webhook 端点作为 POST 端点添加到 FastAPI 路由器应用程序。一旦所有 webhook 都注册完毕,必须调用 `launch` 方法来启动应用程序。

建议将 WebhookPayload 作为 webhook 函数的第一个参数。它是一个 Pydantic 模型,包含有关 webhook 事件的所有信息。数据将自动为您解析。

查看webhooks 指南,获取有关如何设置 WebhooksServer 并将其部署到 Space 的分步教程。

WebhooksServer 是实验性功能。其 API 未来可能会更改。

您必须安装 gradio 才能使用 WebhooksServer (pip install --upgrade gradio)。

示例

import gradio as gr
from huggingface_hub import WebhooksServer, WebhookPayload

with gr.Blocks() as ui:
    ...

app = WebhooksServer(ui=ui, webhook_secret="my_secret_key")

@app.add_webhook("/say_hello")
async def hello(payload: WebhookPayload):
    return {"message": "hello"}

app.launch()

@webhook_endpoint

huggingface_hub.webhook_endpoint

< >

( path: typing.Optional[str] = None )

参数

  • path (str, optional) — 用于注册 webhook 函数的 URL 路径。如果未提供,则函数名称将用作路径。在任何情况下,所有 webhook 都将在 /webhooks 下注册。

用于启动 WebhooksServer() 并将装饰函数注册为 webhook 端点的装饰器。

这是一个快速入门的帮助器。如果您需要更大的灵活性(自定义登录页或 webhook 密钥),可以直接使用 WebhooksServer()。您可以通过多次使用此装饰器来注册多个 webhook 端点(到同一服务器)。

查看webhooks 指南,获取有关如何设置服务器并将其部署到 Space 的分步教程。

webhook_endpoint 是实验性功能。其 API 未来可能会更改。

您必须安装 gradio 才能使用 webhook_endpoint (pip install --upgrade gradio)。

示例:默认用法是将函数注册为 webhook 端点。函数名将用作路径。服务器将在退出时(即脚本结束时)自动启动。

from huggingface_hub import webhook_endpoint, WebhookPayload

@webhook_endpoint
async def trigger_training(payload: WebhookPayload):
    if payload.repo.type == "dataset" and payload.event.action == "update":
        # Trigger a training job if a dataset is updated
        ...

# Server is automatically started at the end of the script.

高级用法:将函数注册为 webhook 端点并手动启动服务器。这在您在笔记本中运行它时很有用。

from huggingface_hub import webhook_endpoint, WebhookPayload

@webhook_endpoint
async def trigger_training(payload: WebhookPayload):
    if payload.repo.type == "dataset" and payload.event.action == "update":
        # Trigger a training job if a dataset is updated
        ...

# Start the server manually
trigger_training.launch()

负载

WebhookPayload 是包含 Webhooks 负载的主要数据结构。这是一个 pydantic 类,使其非常易于与 FastAPI 一起使用。如果将其作为参数传递给 webhook 端点,它将自动验证并解析为 Python 对象。

有关 webhook 负载的更多信息,您可以参考 Webhooks 负载指南

class huggingface_hub.WebhookPayload

< >

( event: WebhookPayloadEvent repo: WebhookPayloadRepo discussion: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadDiscussion] = None comment: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadComment] = None webhook: WebhookPayloadWebhook movedTo: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadMovedTo] = None updatedRefs: typing.Optional[typing.List[huggingface_hub._webhooks_payload.WebhookPayloadUpdatedRef]] = None )

WebhookPayload

class huggingface_hub.WebhookPayload

< >

( event: WebhookPayloadEvent repo: WebhookPayloadRepo discussion: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadDiscussion] = None comment: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadComment] = None webhook: WebhookPayloadWebhook movedTo: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadMovedTo] = None updatedRefs: typing.Optional[typing.List[huggingface_hub._webhooks_payload.WebhookPayloadUpdatedRef]] = None )

WebhookPayloadComment

class huggingface_hub.WebhookPayloadComment

< >

( id: str author: ObjectId hidden: bool content: typing.Optional[str] = None url: WebhookPayloadUrl )

WebhookPayloadDiscussion

class huggingface_hub.WebhookPayloadDiscussion

< >

( id: str num: int author: ObjectId url: WebhookPayloadUrl title: str isPullRequest: bool status: typing.Literal['closed', 'draft', 'open', 'merged'] changes: typing.Optional[huggingface_hub._webhooks_payload.WebhookPayloadDiscussionChanges] = None pinned: typing.Optional[bool] = None )

WebhookPayloadDiscussionChanges

class huggingface_hub.WebhookPayloadDiscussionChanges

< >

( base: str mergeCommitId: typing.Optional[str] = None )

WebhookPayloadEvent

class huggingface_hub.WebhookPayloadEvent

< >

( action: typing.Literal['create', 'delete', 'move', 'update'] scope: str )

WebhookPayloadMovedTo

class huggingface_hub.WebhookPayloadMovedTo

< >

( name: str owner: ObjectId )

WebhookPayloadRepo

class huggingface_hub.WebhookPayloadRepo

< >

( id: str owner: ObjectId head_sha: typing.Optional[str] = None name: str private: bool subdomain: typing.Optional[str] = None tags: typing.Optional[typing.List[str]] = None type: typing.Literal['dataset', 'model', 'space'] url: WebhookPayloadUrl )

WebhookPayloadUrl

class huggingface_hub.WebhookPayloadUrl

< >

( web: str api: typing.Optional[str] = None )

WebhookPayloadWebhook

class huggingface_hub.WebhookPayloadWebhook

< >

( id: str version: typing.Literal[3] )

< > 在 GitHub 上更新