main.py

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}