Spaces:
Paused
Paused
File size: 2,639 Bytes
8035330 253bbca 6ac4ade 8035330 3b22eeb 6d8e1bf 3b22eeb 6d8e1bf c892caa c0cfc84 253bbca bc828c5 8035330 bc828c5 6ac4ade 6ca460a 6ac4ade 8035330 c0cfc84 3b22eeb c892caa 3b22eeb 6d8e1bf 3b22eeb 8035330 3b22eeb 6ca460a 8035330 253bbca 8035330 |
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 |
import contextlib
import json
import os
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, Header
from fastapi.middleware.cors import CORSMiddleware
from huggingface_hub import CommitScheduler, HfApi, whoami
from huggingface_hub.utils._errors import HTTPError
from pydantic import BaseModel, Field
from starlette.responses import RedirectResponse
load_dotenv()
HF_TOKEN = os.getenv("HF_TOKEN")
hf_api = HfApi(token=HF_TOKEN)
scheduler = CommitScheduler(
repo_id="davanstrien/votes",
repo_type="dataset",
folder_path="votes",
path_in_repo="data",
every=1,
token=HF_TOKEN,
hf_api=hf_api,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
Path("votes").mkdir(exist_ok=True)
yield
app = FastAPI()
VOTES_FILE = "votes/votes.jsonl"
# Configure CORS
origins = [
"https://huggingface.co",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["POST"],
allow_headers=["*"],
)
def save_vote(vote_entry):
with open(VOTES_FILE, "a") as file:
# add time stamp to the vote entry
date_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
vote_entry["timestamp"] = date_time
json.dump(vote_entry, file)
file.write("\n")
@app.get("/", include_in_schema=False)
def root():
return RedirectResponse(url="/docs")
class Vote(BaseModel):
dataset: str
description: str
vote: int = Field(..., ge=0, le=1)
userID: str
def validate_token(token: str):
try:
whoami(token=token)
return True
except HTTPError as e:
# check for HTTPError: Invalid user token. If you didn't pass a user token, make sure you are properly logged in by executing `huggingface-cli login`, and if you did pass a user token, double-check it's correct. in the error message
if "Invalid user token" in str(e):
return False
@app.post("/vote")
async def receive_vote(
vote: Vote, background_tasks: BackgroundTasks, authorization: str = Header(...)
):
token = authorization.split(" ")[1]
if not validate_token(token):
# Return a not authorized response
return {"message": "Not authorized"}
vote_entry = {
"dataset": vote.dataset,
"vote": vote.vote,
"description": vote.description,
"userID": vote.userID,
}
# Append the vote entry to the JSONL file
background_tasks.add_task(save_vote, vote_entry)
return {"message": "Vote submitted successfully"}
|