Spaces:
Sleeping
Sleeping
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() | |
def read_root(): | |
return {"Hello": "World!"} | |
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} |