argilla.webhooks

网络钩子是一种 Web 应用程序在发生某些事件时相互通知的方式。例如，您可能希望在新数据集在 Argilla 中创建时收到通知。

使用示例

要监听传入的网络钩子，您可以使用 webhook_listener 装饰器函数来注册一个在接收到 webhook 时调用的函数

from argilla.webhooks import webhook_listener

@webhook_listener(events="dataset.created")
async def my_webhook_listener(dataset):
    print(dataset)

要手动创建新的 webhook，请使用客户端和名称实例化 Webhook 对象

webhook = rg.Webhook(
    url="https://somehost.com/webhook",
    events=["dataset.created"],
    description="My webhook"
)
webhook.create()

要检索现有 webhooks 的列表，请使用 client.webhooks 属性

for webhook in client.webhooks():
    print(webhook)

---

Webhook

基类：Resource

Webhook 资源。它表示一个可用于从 Argilla 服务器接收事件的 webhook。

参数

名称	类型	描述	默认值
url	str	webhook 端点的 URL。	必需
events	List[EventType]	webhook 订阅的事件。	必需
description	Optional[str]	webhook 的描述。	无
_client	Argilla	用于与 Argilla 服务器交互的客户端。	无

源代码位于 src/argilla/webhooks/_resource.py

class Webhook(Resource):
    """
    The `Webhook` resource. It represents a webhook that can be used to receive events from the Argilla Server.

    Args:
        url (str): The URL of the webhook endpoint.
        events (List[EventType]): The events that the webhook is subscribed to.
        description (Optional[str]): The description of the webhook.
        _client (Argilla): The client used to interact with the Argilla Server.

    """

    _model: WebhookModel
    _api: WebhooksAPI

    def __init__(self, url: str, events: List[EventType], description: Optional[str] = None, _client: Argilla = None):
        client = _client or Argilla._get_default()
        api = client.api.webhooks
        events = events or []

        super().__init__(api=api, client=client)

        self._model = WebhookModel(url=url, events=list(events), description=description)

    @property
    def url(self) -> str:
        """The URL of the webhook."""
        return self._model.url

    @url.setter
    def url(self, value: str):
        self._model.url = value

    @property
    def events(self) -> List[EventType]:
        """The events that the webhook is subscribed to."""
        return self._model.events

    @events.setter
    def events(self, value: List[EventType]):
        self._model.events = value

    @property
    def enabled(self) -> bool:
        """Whether the webhook is enabled."""
        return self._model.enabled

    @enabled.setter
    def enabled(self, value: bool):
        self._model.enabled = value

    @property
    def description(self) -> Optional[str]:
        """The description of the webhook."""
        return self._model.description

    @description.setter
    def description(self, value: Optional[str]):
        self._model.description = value

    @property
    def secret(self) -> str:
        """The secret of the webhook."""
        return self._model.secret

    @classmethod
    def from_model(cls, model: WebhookModel, client: Optional["Argilla"] = None) -> "Webhook":
        instance = cls(url=model.url, events=model.events, _client=client)
        instance._model = model

        return instance

    def _with_client(self, client: "Argilla") -> "Webhook":
        self._client = client
        self._api = client.api.webhooks

        return self

url: str property writable

webhook 的 URL。

events: List[EventType] property writable

webhook 订阅的事件。

enabled: bool property writable

webhook 是否启用。

description: Optional[str] property writable

webhook 的描述。

secret: str property

webhook 的密钥。

webhook_listener(events, description=None, client=None, server=None, raw_event=False)

用于为函数创建 webhook 监听器的装饰器。

参数

名称	类型	描述	默认值
events	Union[str, List[str]]	要监听的事件。	必需
description	Optional[str]	webhook 的描述。	无
client	Optional[Argilla]	要使用的 Argilla 客户端。默认为默认客户端。	无
server	Optional[FastAPI]	要使用的 FastAPI 服务器。默认为默认服务器。	无
raw_event	bool	是否将原始事件传递给函数。默认为 False。	False

返回值

名称	类型	描述
Callable	Callable	被装饰的函数。

源代码位于 src/argilla/webhooks/_helpers.py

def webhook_listener(
    events: Union[str, List[str]],
    description: Optional[str] = None,
    client: Optional["Argilla"] = None,
    server: Optional["FastAPI"] = None,
    raw_event: bool = False,
) -> Callable:
    """
    Decorator to create a webhook listener for a function.

    Parameters:
        events (Union[str, List[str]]): The events to listen to.
        description (Optional[str]): The description of the webhook.
        client (Optional[Argilla]): The Argilla client to use. Defaults to the default client.
        server (Optional[FastAPI]): The FastAPI server to use. Defaults to the default server.
        raw_event (bool): Whether to pass the raw event to the function. Defaults to False.

    Returns:
        Callable: The decorated function.

    """

    client = client or rg.Argilla._get_default()
    server = server or get_webhook_server()

    if isinstance(events, str):
        events = [events]

    def wrapper(func: Callable) -> Callable:
        webhook_url = _webhook_url_for_func(func)

        webhook = None
        for argilla_webhook in client.webhooks:
            if argilla_webhook.url == webhook_url and argilla_webhook.events == events:
                warnings.warn(f"Found existing webhook with for URL {argilla_webhook.url}: {argilla_webhook}")
                webhook = argilla_webhook
                webhook.description = description or webhook.description
                webhook.enabled = True
                webhook.update()
                break

        if not webhook:
            webhook = Webhook(
                url=webhook_url,
                events=events,
                description=description or f"Webhook for {func.__name__}",
            ).create()

        request_handler = WebhookHandler(webhook).handle(func, raw_event)
        server.post(f"/{func.__name__}", tags=["Argilla Webhooks"])(request_handler)

        return request_handler

    return wrapper

get_webhook_server()

获取当前的 webhook 服务器。如果它不存在，则会创建一个。

返回值

名称	类型	描述
FastAPI	FastAPI	webhook 服务器。

源代码位于 src/argilla/webhooks/_helpers.py

def get_webhook_server() -> "FastAPI":
    """
    Get the current webhook server. If it does not exist, it will create one.

    Returns:
        FastAPI: The webhook server.

    """
    from fastapi import FastAPI

    global _server
    if not _server:
        _server = FastAPI()
    return _server

set_webhook_server(app)

设置 webhook 服务器。这应该只被调用一次。

参数

名称	类型	描述	默认值
app	FastAPI	webhook 服务器。	必需

源代码位于 src/argilla/webhooks/_helpers.py

def set_webhook_server(app: "FastAPI"):
    """
    Set the webhook server. This should only be called once.

    Parameters:
        app (FastAPI): The webhook server.

    """
    global _server

    if _server:
        raise ValueError("Server already set")

    _server = app

WebhookHandler

WebhookHandler 类用于处理传入的 webhook 请求。此类处理请求验证和事件对象创建。

属性

名称	类型	描述
webhook	Webhook	webhook 对象。

源代码位于 src/argilla/webhooks/_handler.py

class WebhookHandler:
    """
    The `WebhookHandler` class is used to handle incoming webhook requests. This class handles the
    request verification and event object creation.

    Attributes:
        webhook (Webhook): The webhook object.
    """

    def __init__(self, webhook: "Webhook"):
        self.webhook = webhook

    def handle(self, func: Callable, raw_event: bool = False) -> Callable:
        """
        This method handles the incoming webhook requests and calls the provided function.

        Parameters:
            func (Callable): The function to be called when a webhook event is received.
            raw_event (bool): Whether to pass the raw event object to the function.

        Returns:

        """
        from fastapi import Request

        async def request_handler(request: Request):
            event = await self._verify_request(request)
            if event.type not in self.webhook.events:
                return

            if raw_event:
                return await func(event)

            return await func(**event.parsed(self.webhook._client).model_dump())

        return request_handler

    async def _verify_request(self, request: "Request") -> WebhookEvent:
        """
        Verify the request signature and return the event object.

        Arguments:
            request (Request): The request object.

        Returns:
            WebhookEvent: The event object.
        """

        from standardwebhooks.webhooks import Webhook

        body = await request.body()
        headers = dict(request.headers)

        json = Webhook(whsecret=self.webhook.secret).verify(body, headers)
        return WebhookEvent.model_validate(json)

handle(func, raw_event=False)

此方法处理传入的 webhook 请求并调用提供的函数。

参数

名称	类型	描述	默认值
func	Callable	当接收到 webhook 事件时要调用的函数。	必需
raw_event	bool	是否将原始事件对象传递给函数。	False

返回值

源代码位于 src/argilla/webhooks/_handler.py

def handle(self, func: Callable, raw_event: bool = False) -> Callable:
    """
    This method handles the incoming webhook requests and calls the provided function.

    Parameters:
        func (Callable): The function to be called when a webhook event is received.
        raw_event (bool): Whether to pass the raw event object to the function.

    Returns:

    """
    from fastapi import Request

    async def request_handler(request: Request):
        event = await self._verify_request(request)
        if event.type not in self.webhook.events:
            return

        if raw_event:
            return await func(event)

        return await func(**event.parsed(self.webhook._client).model_dump())

    return request_handler

WebhookEvent

基类：BaseModel

一个 webhook 事件。

属性

名称	类型	描述
type	EventType	事件的类型。
timestamp	datetime	事件的时间戳。
data	dict	事件的数据。

源代码位于 src/argilla/webhooks/_event.py

class WebhookEvent(BaseModel):
    """
    A webhook event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        data (dict): The data of the event.
    """

    type: EventType
    timestamp: datetime
    data: dict

    def parsed(self, client: "Argilla") -> Union[RecordEvent, DatasetEvent, UserResponseEvent, "WebhookEvent"]:
        """
        Parse the webhook event.

        Args:
            client: The Argilla client.

        Returns:
            Event: The parsed event.

        """
        resource = self.type.resource
        data = self.data or {}

        if resource == "dataset":
            dataset = self._parse_dataset_from_webhook_data(data, client)
            return DatasetEvent(
                type=self.type,
                timestamp=self.timestamp,
                dataset=dataset,
            )

        elif resource == "record":
            record = self._parse_record_from_webhook_data(data, client)
            return RecordEvent(
                type=self.type,
                timestamp=self.timestamp,
                record=record,
            )

        elif resource == "response":
            user_response = self._parse_response_from_webhook_data(data, client)
            return UserResponseEvent(
                type=self.type,
                timestamp=self.timestamp,
                response=user_response,
            )

        return self

    @classmethod
    def _parse_dataset_from_webhook_data(cls, data: dict, client: "Argilla") -> Dataset:
        workspace = Workspace.from_model(WorkspaceModel.model_validate(data["workspace"]), client=client)
        # TODO: Parse settings from the data
        # settings = Settings._from_dict(data)

        dataset = Dataset(name=data["name"], workspace=workspace, client=client)
        dataset.id = UUID(data["id"])

        try:
            dataset.get()
        except ArgillaAPIError as _:
            # TODO: Show notification
            pass
        finally:
            return dataset

    @classmethod
    def _parse_record_from_webhook_data(cls, data: dict, client: "Argilla") -> Record:
        dataset = cls._parse_dataset_from_webhook_data(data["dataset"], client)

        record = Record.from_model(RecordModel.model_validate(data), dataset=dataset)
        try:
            record.get()
        except ArgillaAPIError as _:
            # TODO: Show notification
            pass
        finally:
            return record

    @classmethod
    def _parse_response_from_webhook_data(cls, data: dict, client: "Argilla") -> UserResponse:
        record = cls._parse_record_from_webhook_data(data["record"], client)

        # TODO: Link the user resource to the response
        user_response = UserResponse.from_model(
            model=UserResponseModel(**data, user_id=data["user"]["id"]),
            record=record,
        )

        return user_response

parsed(client)

解析 webhook 事件。

参数

名称	类型	描述	默认值
client	Argilla	Argilla 客户端。	必需

返回值

名称	类型	描述
Event	Union[RecordEvent, DatasetEvent, UserResponseEvent, WebhookEvent]	解析后的事件。

源代码位于 src/argilla/webhooks/_event.py

def parsed(self, client: "Argilla") -> Union[RecordEvent, DatasetEvent, UserResponseEvent, "WebhookEvent"]:
    """
    Parse the webhook event.

    Args:
        client: The Argilla client.

    Returns:
        Event: The parsed event.

    """
    resource = self.type.resource
    data = self.data or {}

    if resource == "dataset":
        dataset = self._parse_dataset_from_webhook_data(data, client)
        return DatasetEvent(
            type=self.type,
            timestamp=self.timestamp,
            dataset=dataset,
        )

    elif resource == "record":
        record = self._parse_record_from_webhook_data(data, client)
        return RecordEvent(
            type=self.type,
            timestamp=self.timestamp,
            record=record,
        )

    elif resource == "response":
        user_response = self._parse_response_from_webhook_data(data, client)
        return UserResponseEvent(
            type=self.type,
            timestamp=self.timestamp,
            response=user_response,
        )

    return self

DatasetEvent

基类：BaseModel

一个解析后的数据集事件。

属性

名称	类型	描述
type	EventType	事件的类型。
timestamp	datetime	事件的时间戳。
dataset	Dataset	事件的数据集。

源代码位于 src/argilla/webhooks/_event.py

class DatasetEvent(BaseModel):
    """
    A parsed dataset event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        dataset (Dataset): The dataset of the event.
    """

    type: EventType
    timestamp: datetime
    dataset: Dataset

    model_config = ConfigDict(arbitrary_types_allowed=True)

RecordEvent

基类：BaseModel

一个解析后的记录事件。

属性

名称	类型	描述
type	EventType	事件的类型。
timestamp	datetime	事件的时间戳。
record	Record	事件的记录。

源代码位于 src/argilla/webhooks/_event.py

class RecordEvent(BaseModel):
    """
    A parsed record event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        record (Record): The record of the event.
    """

    type: EventType
    timestamp: datetime
    record: Record

    model_config = ConfigDict(arbitrary_types_allowed=True)

UserResponseEvent

基类：BaseModel

一个解析后的用户响应事件。

属性

名称	类型	描述
type	EventType	事件的类型。
timestamp	datetime	事件的时间戳。
response	UserResponse	事件的用户响应。

源代码位于 src/argilla/webhooks/_event.py

class UserResponseEvent(BaseModel):
    """
    A parsed user response event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        response (UserResponse): The user response of the event.
    """

    type: EventType
    timestamp: datetime
    response: UserResponse

    model_config = ConfigDict(arbitrary_types_allowed=True)