gradio_webhooks.py

from pathlib import Path
from typing import Set, Union

import gradio as gr


class WebhookGradioApp:
    """
    ```py
    from gradio_webhooks import WebhookGradioApp

    app = WebhookGradioApp()


    @app.add_webhook("/test_webhook")
    async def hello():
        return {"in_gradio": True}


    app.block_thread()
    ```
    """

    def __init__(self, landing_path: Union[str, Path] = "README.md") -> None:
        # Use README.md as landing page or provide any markdown file
        landing_path = Path(landing_path)
        landing_content = landing_path.read_text()
        if landing_path.name == "README.md":
            landing_content = landing_content.split("---")[-1].strip()

        # Simple gradio app with landing content
        block = gr.Blocks()
        with block:
            gr.Markdown(landing_content)

        # Launch gradio app:
        #   - as non-blocking so that webhooks can be added afterwards
        #   - as shared if launch locally (to receive webhooks)
        app, _, _ = block.launch(prevent_thread_lock=True, share=not block.is_space)
        self.gradio_app = block
        self.fastapi_app = app
        self.webhook_paths: Set[str] = set()

    def add_webhook(self, path: str):
        self.webhook_paths.add(path)
        return self.fastapi_app.post(path)

    def block_thread(self) -> None:
        url = (
            self.gradio_app.share_url
            if self.gradio_app.share_url is not None
            else self.gradio_app.local_url
        ).strip("/")
        print("\nWebhooks are correctly setup and ready to use:")
        print("\n".join(f"  - POST {url}{webhook}" for webhook in self.webhook_paths))
        print(f"Checkout {url}/docs for more details.")
        self.gradio_app.block_thread()