File size: 21,905 Bytes
0fc235e
 
 
fd92af7
ac36ca5
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
fd92af7
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ac36ca5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3855dd6
 
ac36ca5
3855dd6
 
ac36ca5
 
 
 
 
 
 
3855dd6
 
 
 
ac36ca5
 
3855dd6
ac36ca5
 
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
fd92af7
f9a117c
 
 
0fc235e
 
 
fd92af7
 
 
 
 
 
 
 
 
 
 
9bbfce5
de95380
 
 
 
 
 
 
 
 
 
 
 
fd92af7
 
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
fd92af7
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
fd92af7
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
fd92af7
0fc235e
f9a117c
 
 
 
0fc235e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
import os
import uuid
import warnings
from concurrent.futures import ThreadPoolExecutor
from functools import wraps
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,
    repo_exists,
    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 enable_space_ci() -> None:
    """Enable Space CI for the current Space based on config from the README.md file.

    Example:
    ```py
    import gradio as gr
    from gradio_space_ci import enable_space_ci

    enable_space_ci()

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

    demo.launch()
    ```
    """
    if SPACE_ID is None:
        print("Not in a Space: Space CI disabled.")
        return

    if IS_EPHEMERAL_SPACE:
        print("In an ephemeral Space: Space CI disabled.")
        return

    card = RepoCard.load(repo_id_or_path=SPACE_ID, repo_type="space")
    config = card.data.get("space_ci", {})
    print(f"Enabling Space CI with config from README: {config}")

    old_launch = gr.Blocks.launch

    @wraps(gr.Blocks.launch)
    def new_launch(self: gr.Blocks, *args, **kwargs) -> None:
        server = configure_space_ci(
            blocks=self,
            trusted_authors=config.get("trusted_authors"),
            private=config.get("private", "auto"),
            variables=config.get("variables", "auto"),
            secrets=config.get("secrets"),
            hardware=config.get("hardware"),
            storage=config.get("storage"),
        )
        # De-monkey patch gradio (otherwise it will be called recursively)
        gr.Blocks.launch = old_launch
        server.launch(*args, *kwargs)

    # Monkey patch gradio
    gr.Blocks.launch = new_launch


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()

    # Recover missed webhooks (loop through PRs in the background)
    background_pool.submit(recover_after_restart, space_id=SPACE_ID)

    return server


###
# Recovery logic
###
# Check if there are any PRs that need to be synced.
# We might have missed some events while the server was down.
# => called once at startup (see configure_space_ci)

background_pool = ThreadPoolExecutor(max_workers=1)


def recover_after_restart(space_id: str) -> None:
    print("Looping through PRs to check if any needs to be synced.")
    for discussion in get_repo_discussions(repo_id=space_id, repo_type="space", discussion_type="pull_request"):
        if discussion.status == "open":
            if not is_pr_synced(space_id=space_id, pr_num=discussion.num):
                # Found a PR that is not yet synced
                print(f"Recovery. Found an open PR that is not synced: {discussion.url}. Syncing it.")
                background_pool.submit(sync_ci_space, space_id=space_id, pr_num=discussion.num)
        if discussion.status == "merged" or discussion.status == "closed":
            ci_space_id = _get_ci_space_id(space_id=space_id, pr_num=discussion.num)
            if repo_exists(repo_id=ci_space_id, repo_type="space"):
                # Found a PR for which the CI space has not been deleted
                print(f"Recovery. Found a closed PR with an active CI space: {discussion.url}. Deleting it.")
                background_pool.submit(delete_ci_space, space_id=space_id, pr_num=discussion.num)


###
# 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)
            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)
                    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) -> None:
    print(f"New task: sync ephemeral env for {space_id} (PR {pr_num})")
    if is_pr_synced(space_id=space_id, pr_num=pr_num):
        print("Already synced. Nothing to do.")
        return

    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)