main.py

import os
from queue import Queue

import argilla as rg
import gradio as gr

client = rg.Argilla()

server = rg.get_webhook_server()
incoming_events = Queue()


# Set up the webhook listeners

# Create a webhook for record events

@rg.webhook_listener(events=["record.created", "record.updated", "record.completed"])
async def record_events(record: rg.Record, type: str, **kwargs):
    print("Received event", type, record)

    incoming_events.put(record)


# Create a webhook for dataset events
@rg.webhook_listener(events=["dataset.created", "dataset.updated", "dataset.published"])
async def dataset_events(type: str, dataset: rg.Dataset, **kwargs):
    print(f"Received event", type, dataset)

    incoming_events.put((type, dataset))


# Create a webhook for response events
@rg.webhook_listener(events=["response.created", "response.updated"])
async def response_events(type:str, response: rg.UserResponse,  **kwargs):
    print(f"Received event", type, response)

    incoming_events.put(response)


@rg.webhook_listener(events=["record.deleted", "dataset.deleted", "response.deleted"])
async def deleted_events(type: str, data: dict, **kwargs):
    print(f"Received event {type} for resource {data}")

    incoming_events.put((type, data))


def check_incoming_events():
    """
    This function is called every 5 seconds to check if there are any incoming
    events and send data to update the JSON component.
    """
    event = incoming_events.get()

    return event

with gr.Blocks() as demo:
    argilla_server = client.http_client.base_url
    gr.Markdown("## Argilla Events")
    gr.Markdown(f"This demo shows the incoming events from the [Argilla Server]({argilla_server}).")
    json_component = gr.JSON(label="Incoming argilla events:")
    gr.Timer(1, active=True).tick(check_incoming_events, outputs=json_component)

gr.mount_gradio_app(server, demo, path="/")