Spaces:
Sleeping
Sleeping
File size: 1,601 Bytes
b9721b4 23b4bde de969c6 b9721b4 ba0c441 23b4bde b9721b4 fdd3a00 b9721b4 fdd3a00 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 |
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
snapshot_download(repo_id="SakethTest/ThirdParty",local_dir="./ThirdParty")
try:
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} |