Spaces:
Sleeping
Sleeping
File size: 1,673 Bytes
b9721b4 23b4bde de969c6 b9721b4 ba0c441 23b4bde b9721b4 678a8d5 b9721b4 de969c6 b9721b4 |
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 |
import os
import requests
from typing import Optional
from huggingface_hub import snapshot_download
from fastapi import FastAPI, Header, HTTPException, BackgroundTasks
from huggingface_hub.hf_api import HfApi
from .models import config, WebhookPayload
app = FastAPI()
api = HfApi()
@app.get("/")
def read_root():
return {"Hello": "World!"}
@app.post("/webhook")
async def post_webhook(
payload: WebhookPayload,
task_queue: BackgroundTasks,
x_webhook_secret: Optional[str] = Header(default=None),
):
if x_webhook_secret is None:
raise HTTPException(401)
if x_webhook_secret != "webhooksecret":
raise HTTPException(403)
if not (
payload.event.action == "update"
and payload.event.scope.startswith("repo.content")
and payload.repo.name == "SakethTest/ThirdParty"
and payload.repo.type == "model"
):
# no-op
return {"processed": False}
task_queue.add_task(
update_cloned_repo,
payload
)
return {"processed": True}
def update_cloned_repo(payload: WebhookPayload):
# Create the update_cloned_repo project
try:
snapshot_download(repo_id="SakethTest/ThirdParty",local_dir="./ThirdParty")
api.upload_folder(
folder_path="./ThirdParty",
repo_id="shellplc/ThirdParty",
repo_type="model",
commit_message="uploaded third party model",
token="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"
)
except requests.HTTPError as err:
print("ERROR while requesting AutoTrain API:")
print(f" code: {err.response.status_code}")
print(f" {err.response.json()}")
raise
# Notify in the community tab
notify_success(project["id"])
return {"processed": True} |