first commit

.gitignore

@@ -0,0 +1,144 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+.venv*
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+.venv*
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+.vscode/
+.idea/
+
+.DS_Store
+
+# Ruff
+.ruff_cache
+
+# Spell checker config
+cspell.json
+
+mock
+_user_history
+_user_history_exports

Makefile

@@ -0,0 +1,10 @@
+.PHONY: quality style
+
+quality:
+	ruff format . --check
+	ruff check .
+	mypy --install-types .
+
+style:
+	ruff format .
+	ruff check --fix .

README.md

@@ -0,0 +1,92 @@
+---
+title: Gradio Space CI
+emoji: 🤖
+colorFrom: gray
+colorTo: gray
+sdk: gradio
+sdk_version: 4.7.1
+app_file: app.py
+pinned: false
+---
+
+# Build a CI for your Spaces 🚀
+
+**Gradio Space CI** is a plugin (and package) to create ephemeral Spaces for each PR opened on your Space repo.
+The goal is to foster community contributions by making the review process as lean as possible.
+
+## Key features
+
+1. **Listen to Pull Requests**.
+2. **Launch ephemeral Spaces** on new PRs.
+   1. When a commit is pushed to a PR, the ephemeral Space gets synchronized.
+   2. When the PR is closed, the ephemeral Space is deleted.
+3. Configure ephemeral Spaces automatically
+   1. All **variables** are copied from the main Space.
+   2. **Secrets** are copied from the main Space, based on CI configuration.
+   3. **Hardware** and **storage** are set, based on CI configuration.
+4. Only **trusted authors** are able to access secrets
+   1. By default, repo owners are trusted authors
+   2. More authors can be added in CI configuration
+   3. untrusted authors can start ephemeral Space but without secrets or custom hardware
+
+Want more? Please open an issue in the [Community Tab](https://huggingface.co/spaces/Wauplin/gradio-space-ci/discussions)! This is meant to be a community-driven implementation, enhanced by user feedback and contributions!
+
+## Integration
+
+Integrate *Gradio Space CI* in just a few steps:
+
+### 1. Update your `requirements.txt`
+
+If you don't have a `requirements.txt` file yet, create one in your Space repo.
+Add the following line to it:
+
+```bash
+# requirements.txt
+git+https://huggingface.co/spaces/Wauplin/gradio-space-ci
+```
+
+
+### 2. Add a user token as `HF_TOKEN` secret
+
+1. Go to your [user settings page](https://huggingface.co/settings/tokens).
+2. Create a new token with **write** permissions.
+3. Go to your Space settings page.
+4. Add `HF_TOKEN` as a Space secret with your newly created token.
+
+(optional) You can also define a `SPACE_CI_SECRET` secret value that will be used to authenticate webhook calls. If you
+don't define one, a random secret value will be generated for you.
+
+
+### 3. Configure CI in `app.py`
+
+Import the `gradio_space_ci` package and update the last line of your `app.py` script, just before launching the Gradio app.
+
+```py
+# app.py
+import gradio as gr
+from gradio_space_ci import configure_space_ci
+
+# ANY gradio app
+with gr.Blocks() as demo:
+    ...
+
+# Replace `demo.launch()` by `configure_space_ci(demo).launch()`.
+configure_space_ci(
+    demo,
+    trusted_authors=[],  # space owners + manually trusted authors
+    private="auto",  # ephemeral spaces will have same visibility as the main space. Otherwise, set to `True` or `False` explicitly.
+    variables="auto",  # same variables as the main space. Otherwise, set to a `Dict[str, str]`.
+    secrets=["HF_TOKEN"],  # which secret do I want to copy from the main space? Can be a `List[str]`.
+    hardware=None,  # "cpu-basic" by default. Otherwise set to "auto" to have same hardware as the main space or any valid string value.
+    storage=None,  # no storage by default. Otherwise set to "auto" to have same storage as the main space or any valid string value.
+).launch()
+```
+
+And you're done! Ephemeral Spaces will be launched for each and every PR on your repo.
+
+
+## Useful links
+
+- **Demo:** https://huggingface.co/spaces/Wauplin/gradio-space-ci
+- **README:** https://huggingface.co/spaces/Wauplin/gradio-space-ci/blob/main/README.md
+- **Questions and feedback:** https://huggingface.co/spaces/Wauplin/gradio-space-ci/discussions

app.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+from pathlib import Path
+
+import gradio as gr
+from gradio_space_ci import configure_space_ci
+
+
+def greet(name: str) -> str:
+    return "Hello " + name + "!"
+
+
+with gr.Blocks() as demo:
+    gr.Markdown("## Dummy gradio app to showcase gradio-space-ci")
+    name = gr.Textbox(label="Name")
+    output = gr.Textbox(label="Output Box")
+    greet_btn = gr.Button("Greet")
+    greet_btn.click(fn=greet, inputs=name, outputs=output, api_name="greet")
+
+
+with gr.Blocks() as demo_with_readme:
+    with gr.Tab("README"):
+        gr.Markdown(Path("README.md").read_text().split("---")[-1])
+    with gr.Tab("Demo"):
+        demo.render()
+
+if __name__ == "__main__":
+    configure_space_ci(
+        blocks=demo_with_readme.queue(),  # ANY gradio app
+        trusted_authors=["clefourrier"],  # space owners + manually trusted authors
+        private="auto",  # ephemeral spaces will have same visibility as the main space. Otherwise, set to `True` or `False` explicitly.
+        variables="auto",  # same variables as the main space. Otherwise, set to a `Dict[str, str]`.
+        secrets=["HF_TOKEN"],  # which secret do I want to copy from the main space? Can be a `List[str]`.
+        hardware=None,  # "cpu-basic" by default. Otherwise set to "auto" to have same hardware as the main space or any valid string value.
+        storage=None,  # no storage by default. Otherwise set to "auto" to have same storage as the main space or any valid string value.
+    ).launch()

pyproject.toml

@@ -0,0 +1,19 @@
+[tool.ruff]
+# Enable pycodestyle (`E`) and Pyflakes (`F`) codes by default.
+select = ["E", "F"]
+ignore = ["E501"] # line too long (black is taking care of this)
+line-length = 119
+fixable = ["A", "B", "C", "D", "E", "F", "G", "I", "N", "Q", "S", "T", "W", "ANN", "ARG", "BLE", "COM", "DJ", "DTZ", "EM", "ERA", "EXE", "FBT", "ICN", "INP", "ISC", "NPY", "PD", "PGH", "PIE", "PL", "PT", "PTH", "PYI", "RET", "RSE", "RUF", "SIM", "SLF", "TCH", "TID", "TRY", "UP", "YTT"]
+
+[tool.isort]
+profile = "black"
+line_length = 119
+
+[tool.black]
+line-length = 119
+
+[tool.mypy]
+ignore_missing_imports = true
+no_implicit_optional = true
+scripts_are_modules = true
+

requirements-dev.txt

@@ -0,0 +1,4 @@
+mypy
+ruff
+types-requests
+types-ujson

requirements.txt

@@ -0,0 +1,3 @@
+gradio
+# TODO: update once merged
+git+https://github.com/huggingface/huggingface_hub@main

setup.py

@@ -0,0 +1,57 @@
+from setuptools import find_packages, setup
+
+
+def get_version() -> str:
+    rel_path = "src/gradio_space_ci/__init__.py"
+    with open(rel_path, "r") as fp:
+        for line in fp.read().splitlines():
+            if line.startswith("__version__"):
+                delim = '"' if '"' in line else "'"
+                return line.split(delim)[1]
+    raise RuntimeError("Unable to find version string.")
+
+
+install_requires = [
+    "gradio[oauth]>=3.44",
+]
+
+extras = {}
+
+extras["dev"] = [
+    "ruff",
+    "black",
+    "mypy",
+]
+
+
+setup(
+    name="gradio_space_ci",
+    version=get_version(),
+    author="Lucain Pouget",
+    author_email="[email protected]",
+    description="A package to enable Space CI (ephemeral Spaces on when PR is created).",
+    long_description=open("README.md", "r", encoding="utf-8").read(),
+    long_description_content_type="text/markdown",
+    keywords="gradio spaces ci machine-learning",
+    license="Apache",
+    url="https://huggingface.co/spaces/Wauplin/gradio-space-ci",
+    package_dir={"": "src"},
+    packages=find_packages("src"),
+    extras_require=extras,
+    python_requires=">=3.8.0",
+    install_requires=install_requires,
+    classifiers=[
+        "Intended Audience :: Developers",
+        "Intended Audience :: Education",
+        "Intended Audience :: Science/Research",
+        "License :: OSI Approved :: Apache Software License",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3 :: Only",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Topic :: Scientific/Engineering :: Artificial Intelligence",
+    ],
+)

src/gradio_space_ci/__init__.py

@@ -0,0 +1,40 @@
+"""
+Gradio Space CI is a plugin (and package) to create ephemeral Spaces for each PR opened on your Space repo.
+The goal is to foster community contributions by making the review process as lean as possible.
+
+# Key features:
+1. Listen to Pull Requests.
+2. Launch ephemeral Spaces on new PRs.
+   1. When a commit is pushed to a PR, the ephemeral Space gets synchronized.
+   2. When the PR is closed, the ephemeral Space is deleted.
+3. Configure ephemeral Spaces automatically
+   1. All variables are copied from the main Space.
+   2. Secrets are copied from the main Space, based on CI configuration.
+   3. Hardware and storage are set, based on CI configuration.
+4. Only trusted authors are able to access secrets
+   1. By default, repo owners are trusted authors
+   2. More authors can be added in CI configuration
+   3. untrusted authors can start ephemeral Space but without secrets or custom hardware
+
+# Useful links:
+- Demo: https://huggingface.co/spaces/Wauplin/gradio-space-ci
+- README: https://huggingface.co/spaces/Wauplin/gradio-space-ci/blob/main/README.md
+- Questions and feedback: https://huggingface.co/spaces/Wauplin/gradio-space-ci/discussions
+"""
+import warnings
+
+from huggingface_hub import HfFolder
+
+# Check if `HF_TOKEN` is set. If not, Space CI will be disabled but no error is raised.
+if HfFolder.get_token() is None:
+    warnings.warn(
+        "Cannot find `HF_TOKEN` in environment variables. Please set a token in your Space secrets to enable ephemeral Spaces."
+    )
+
+    def configure_space_ci(blocks, *args, kwargs):
+        return blocks
+else:
+    from .webhook import configure_space_ci  # noqa: F401
+
+
+__version__ = "0.1.0"

src/gradio_space_ci/webhook.py

@@ -0,0 +1,526 @@
+import os
+import uuid
+import warnings
+from pathlib import Path
+from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
+
+import gradio as gr
+from fastapi import BackgroundTasks, HTTPException, Response, status
+from huggingface_hub import (
+    SpaceHardware,
+    SpaceStorage,
+    WebhookPayload,
+    WebhooksServer,
+    add_space_secret,
+    add_space_variable,
+    comment_discussion,
+    create_repo,
+    delete_repo,
+    get_discussion_details,
+    get_repo_discussions,
+    get_space_runtime,
+    get_space_variables,
+    request_space_hardware,
+    request_space_storage,
+    snapshot_download,
+    space_info,
+    upload_folder,
+)
+from huggingface_hub.repocard import RepoCard
+from huggingface_hub.utils import (
+    RepositoryNotFoundError,
+    build_hf_headers,
+    get_session,
+    hf_raise_for_status,
+)
+from requests import HTTPError
+
+SPACE_ID = os.environ.get("SPACE_ID")
+IS_EPHEMERAL_SPACE = SPACE_ID is not None and "-ci-pr-" in SPACE_ID
+WEBHOOK_SECRET = os.environ.get("SPACE_CI_SECRET")
+
+if SPACE_ID is not None:  # If running in a Space (i.e. not locally)
+    if WEBHOOK_SECRET is None:  # No secret set yet => generate one => restart space
+        WEBHOOK_SECRET = str(uuid.uuid4())
+        add_space_secret(
+            repo_id=SPACE_ID,
+            key="SPACE_CI_SECRET",
+            value=WEBHOOK_SECRET,
+            description="This value is used by the SpaceCI. It is automatically generated and should not be changed.",
+        )
+
+EPHEMERAL_SPACES_CONFIG: Dict[str, Any] = {}
+
+
+def configure_space_ci(
+    blocks: Optional["gr.Blocks"] = None,
+    trusted_authors: Optional[List[str]] = None,
+    private: Union[bool, Literal["auto"]] = "auto",
+    variables: Union[Dict[str, str], Literal["auto"]] = "auto",
+    secrets: Optional[List[str]] = None,
+    hardware: Union[SpaceHardware, Literal["auto"], None] = None,
+    storage: Union[SpaceStorage, Literal["auto"], None] = None,
+) -> WebhooksServer:
+    if SPACE_ID is None or IS_EPHEMERAL_SPACE:
+        # Runs locally => don't configure webhook
+        # Runs in an ephemeral Space => don't configure webhook
+        return WebhooksServer(ui=blocks)
+
+    # Authors
+    trusted_authors = trusted_authors or []
+    namespace = SPACE_ID.split("/")[0]
+    try:  # Check if namespace is an organization => in this case all members are allowed to trigger CI by default
+        response = get_session().get(
+            f"https://huggingface.co/api/organizations/{namespace}/members", headers=build_hf_headers()
+        )
+        response.raise_for_status()
+        trusted_authors += [user["user"] for user in response.json()]
+    except Exception:  # Otherwise, it's a single user => only this user is allowed to trigger CI by default
+        trusted_authors += [namespace]
+    trusted_authors = sorted(set(trusted_authors))
+    EPHEMERAL_SPACES_CONFIG["trusted_authors"] = trusted_authors
+
+    # Private
+    if private == "auto":
+        private = space_info(SPACE_ID).private
+    EPHEMERAL_SPACES_CONFIG["private"] = private
+
+    # Variables
+    if variables == "auto":
+        variables = {value.key: value.value for value in get_space_variables(SPACE_ID).values()}
+    EPHEMERAL_SPACES_CONFIG["variables"] = variables
+
+    # Secrets
+    secrets_with_values: Dict[str, str] = {}
+    if secrets is not None:
+        for secret in secrets:
+            secret_value = os.environ.get(secret)
+            if secret_value is None:
+                warnings.warn(f"Secret {secret} not found in environment variables. Will skip it in ephemeral Space.")
+                continue
+            secrets_with_values[secret] = secret_value
+    EPHEMERAL_SPACES_CONFIG["secrets"] = secrets_with_values
+
+    # Hardware and storage
+    if hardware == "auto" or storage == "auto":
+        runtime = get_space_runtime(SPACE_ID)
+        if hardware == "auto":
+            hardware = runtime.hardware
+        if storage == "auto":
+            storage = runtime.storage
+    EPHEMERAL_SPACES_CONFIG["hardware"] = hardware
+    EPHEMERAL_SPACES_CONFIG["storage"] = storage
+
+    # Summary
+    print(
+        "Ephemeral Spaces config:"
+        f"\n  - trusted authors: {trusted_authors}"
+        f"\n  - private: {private}"
+        f"\n  - secrets: {', '.join(sorted(secrets_with_values.keys()))}"
+        f"\n  - variables: {variables}"
+        f"\n  - storage: {storage}"
+        f"\n  - hardware: {hardware}"
+    )
+
+    # Configure webhook
+    server = WebhooksServer(ui=blocks, webhook_secret=WEBHOOK_SECRET)
+    server.add_webhook()(trigger_ci_on_pr)
+    configure_webhook_on_hub()
+    return server
+
+
+###
+# Define webhook on the Hub logic
+###
+
+
+def configure_webhook_on_hub():
+    url = "https://" + os.environ.get("SPACE_HOST").strip("/") + "/webhooks/trigger_ci_on_pr"
+
+    # Check if webhook already exists
+    webhooks = list_webhooks()
+    for webhook in webhooks:
+        if webhook["url"] == url:
+            print("Webhook already configured")
+            return
+
+    # If not => create it
+    create_webhook(
+        watched=[{"type": "space", "name": SPACE_ID}], url=url, domains=["repo", "discussion"], secret=WEBHOOK_SECRET
+    )
+    print("New webhook already configured!")
+
+
+###
+# Webhook logic
+###
+
+
+async def trigger_ci_on_pr(payload: WebhookPayload, task_queue: BackgroundTasks):
+    if payload.repo.type != "space":
+        raise HTTPException(400, f"Must be a Space, not {payload.repo.type}")
+
+    space_id = payload.repo.name
+
+    has_task = False
+    if (
+        # Means "a new PR has been opened"
+        payload.event.scope.startswith("discussion")
+        and payload.event.action == "create"
+        and payload.discussion is not None
+        and payload.discussion.isPullRequest
+        and payload.discussion.status == "open"
+    ):
+        if not is_pr_synced(space_id=space_id, pr_num=payload.discussion.num):
+            # New PR! Sync task scheduled
+            task_queue.add_task(
+                sync_ci_space,
+                space_id=space_id,
+                pr_num=payload.discussion.num,
+                private=payload.repo.private,
+            )
+            has_task = True
+    elif (
+        # Means "a PR has been merged or closed"
+        payload.event.scope.startswith("discussion")
+        and payload.event.action == "update"
+        and payload.discussion is not None
+        and payload.discussion.isPullRequest
+        and (payload.discussion.status == "merged" or payload.discussion.status == "closed")
+    ):
+        task_queue.add_task(
+            delete_ci_space,
+            space_id=space_id,
+            pr_num=payload.discussion.num,
+        )
+        has_task = True
+    elif (
+        # Means "some content has been pushed to the Space" (any branch)
+        payload.event.scope.startswith("repo.content") and payload.event.action == "update"
+    ):
+        # New repo change. Is it a commit on a PR?
+        # => loop through all PRs and check if new changes happened
+        for discussion in get_repo_discussions(repo_id=space_id, repo_type="space"):
+            if discussion.is_pull_request and discussion.status == "open":
+                if not is_pr_synced(space_id=space_id, pr_num=discussion.num):
+                    # Found a PR that is not yet synced
+                    task_queue.add_task(
+                        sync_ci_space,
+                        space_id=space_id,
+                        pr_num=discussion.num,
+                        private=payload.repo.private,
+                    )
+                    has_task = True
+
+    if has_task:
+        return Response("Task scheduled to sync/delete Space", status_code=status.HTTP_202_ACCEPTED)
+    else:
+        return Response("No task scheduled", status_code=status.HTTP_200_OK)
+
+
+###
+# Internal logic
+###
+
+
+def is_pr_synced(space_id: str, pr_num: int) -> bool:
+    # What is the last synced commit for this PR?
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    try:
+        card = RepoCard.load(repo_id_or_path=ci_space_id, repo_type="space")
+        last_synced_sha = getattr(card.data, "synced_sha", None)
+    except HTTPError:
+        return False
+
+    # What is the last commit id for this PR?
+    info = space_info(repo_id=space_id, revision=f"refs/pr/{pr_num}")
+    last_pr_sha = info.sha
+
+    # Is it up to date ?
+    return last_synced_sha == last_pr_sha
+
+
+def sync_ci_space(space_id: str, pr_num: int, private: bool) -> None:
+    print(f"New task: sync ephemeral env for {space_id} (PR {pr_num})")
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+
+    # Create a temporary space for CI if didn't exist
+    is_new = create_ephemeral_space(space_id=space_id, pr_num=pr_num)
+
+    # Configure ephemeral Space if trusted author
+    is_configured = False
+    if is_new:
+        is_configured = configure_ephemeral_space(space_id=space_id, pr_num=pr_num)
+
+    # Download space codebase from PR revision
+    snapshot_path = Path(snapshot_download(repo_id=space_id, revision=f"refs/pr/{pr_num}", repo_type="space"))
+
+    # Overwrite README file in cache (/!\)
+    readme_path = snapshot_path / "README.md"
+    card = RepoCard.load(readme_path)
+    setattr(card.data, "synced_sha", snapshot_path.name)  # latest sha
+    card.data.title = f"{card.data.title} (ephemeral #{pr_num})"
+    card.save(readme_path)
+
+    # Sync space codebase with PR revision
+    upload_folder(
+        repo_id=ci_space_id,
+        repo_type="space",
+        commit_message=f"Sync CI Space with PR {pr_num}.",
+        folder_path=snapshot_path,
+        delete_patterns="*",
+    )
+
+    # Delete readme file from cache (just in case)
+    readme_path.unlink(missing_ok=True)
+
+    # Post a comment on the PR
+    if is_new and is_configured:
+        notify_pr(space_id=space_id, pr_num=pr_num, action="created_and_configured")
+    elif is_new:
+        notify_pr(space_id=space_id, pr_num=pr_num, action="created_not_configured")
+    else:
+        notify_pr(space_id=space_id, pr_num=pr_num, action="updated")
+
+
+def create_ephemeral_space(space_id: str, pr_num: int) -> bool:
+    # Config values
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    private: bool = EPHEMERAL_SPACES_CONFIG["private"]
+
+    # Create space
+    try:
+        create_repo(
+            ci_space_id,
+            repo_type="space",
+            space_sdk="docker",  # Will be overwritten by sync
+            private=private,
+            exist_ok=False,
+        )
+        return True
+    except HTTPError as err:
+        if err.response is not None and err.response.status_code == 409:  # already exists
+            return False
+        else:
+            raise
+
+
+def configure_ephemeral_space(space_id: str, pr_num: int) -> bool:
+    # Config values
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    trusted_authors: List[str] = EPHEMERAL_SPACES_CONFIG["trusted_authors"]
+    variables: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["variables"]
+    secrets: Dict[str, str] = EPHEMERAL_SPACES_CONFIG["secrets"]
+    hardware: Optional[SpaceHardware] = EPHEMERAL_SPACES_CONFIG["hardware"]
+    storage: Optional[SpaceHardware] = EPHEMERAL_SPACES_CONFIG["storage"]
+
+    # Check if trusted author
+    details = get_discussion_details(repo_id=space_id, repo_type="space", discussion_num=pr_num)
+    if details.author not in trusted_authors:
+        return False  # not a trusted author => do NOT set secrets, hardware, storage, etc.
+
+    # Configure space
+    for key, value in variables.items():
+        add_space_variable(ci_space_id, key, value)
+    for key, value in secrets.items():
+        add_space_secret(ci_space_id, key, value)
+
+    # Request hardware/storage for space
+    if hardware is not None and hardware != SpaceHardware.CPU_BASIC:
+        request_space_hardware(ci_space_id, hardware, sleep_time=5 * 60)  # sleep after 5min on PR Spaces with GPU
+    if storage is not None:
+        request_space_storage(ci_space_id, storage)
+
+    return True
+
+
+def delete_ci_space(space_id: str, pr_num: int) -> None:
+    print(f"New task: delete ephemeral env for {space_id} (PR {pr_num})")
+
+    # Delete
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    try:
+        delete_repo(repo_id=ci_space_id, repo_type="space")
+    except RepositoryNotFoundError:
+        # Repo did not exist: no need to notify
+        return
+
+    # Notify about deletion
+    notify_pr(space_id=space_id, pr_num=pr_num, action="deleted")
+
+
+def notify_pr(
+    space_id: str,
+    pr_num: int,
+    action: Literal["created_not_configured", "created_and_configured", "updated", "deleted"],
+) -> None:
+    ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=pr_num)
+    if action == "created_not_configured":
+        comment = NOTIFICATION_TEMPLATE_CREATED_NOT_CONFIGURED.format(ci_space_id=ci_space_id)
+    elif action == "created_and_configured":
+        comment = NOTIFICATION_TEMPLATE_CREATED_AND_CONFIGURED.format(ci_space_id=ci_space_id)
+    elif action == "updated":
+        comment = NOTIFICATION_TEMPLATE_UPDATED.format(ci_space_id=ci_space_id)
+    elif action == "deleted":
+        comment = NOTIFICATION_TEMPLATE_DELETED
+    else:
+        raise ValueError(f"Status {action} not handled.")
+
+    comment_discussion(repo_id=space_id, repo_type="space", discussion_num=pr_num, comment=comment)
+
+
+def _get_ci_space_id(space_id: str, pr_num: int) -> str:
+    return f"{space_id}-ci-pr-{pr_num}"
+
+
+NOTIFICATION_TEMPLATE_CREATED_AND_CONFIGURED = """\
+Following the creation of this PR, an ephemeral Space [{ci_space_id}](https://huggingface.co/spaces/{ci_space_id}) has been started. Any changes pushed to this PR will be synced with the test Space.
+Since this PR has been created by a trusted author, the ephemeral Space has been configured with the correct hardware, storage, and secrets.
+_(This is an automated message.)_
+"""
+
+NOTIFICATION_TEMPLATE_CREATED_NOT_CONFIGURED = """\
+Following the creation of this PR, an ephemeral Space [{ci_space_id}](https://huggingface.co/spaces/{ci_space_id}) has been started. Any changes pushed to this PR will be synced with the test Space.
+Since this PR has not been created by a trusted author, the ephemeral Space has not been configured with the correct hardware, storage, and secrets. An admin must configure it manually.
+_(This is an automated message.)_
+"""
+
+NOTIFICATION_TEMPLATE_UPDATED = """\
+Following new commits that happened in this PR, the ephemeral Space [{ci_space_id}](https://huggingface.co/spaces/{ci_space_id}) has been updated.
+_(This is an automated message.)_
+"""
+
+NOTIFICATION_TEMPLATE_DELETED = """\
+PR is now merged/closed. The ephemeral Space has been deleted.
+_(This is an automated message.)_
+"""
+
+### TO MOVE TO ITS OWN MODULE
+# Taken from https://github.com/huggingface/huggingface_hub/issues/1808#issuecomment-1802341663
+
+
+headers = build_hf_headers()
+
+
+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,
+            },
+        }
+        ```
+    """
+    print("Creating webhook")
+    print({"watched": watched, "url": url, "domains": domains, "secret": str(type(secret))})
+
+    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)