Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10proposals = {}
11votes = {}
12next_user_id = 1
13next_proposal_id = 1
14next_vote_id = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19    track: Optional[str] = None
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class ProposalCreate(BaseModel):
26    title: str
27    abstract: str
28    track: str
29    speaker: str
30 
31class VoteCreate(BaseModel):
32    proposal_id: int
33    score: int
34 
35def get_current_user(token: str = Header(...)):
36    user_id = tokens.get(token)
37    if user_id is None:
38        raise HTTPException(status_code=401, detail="Invalid token")
39    return user_id
40 
41@app.post("/signup")
42def signup(req: SignupRequest):
43    global next_user_id
44    for u in users.values():
45        if u["username"] == req.username:
46            raise HTTPException(status_code=400, detail="Username taken")
47    user_id = next_user_id
48    next_user_id += 1
49    users[user_id] = {"id": user_id, "username": req.username, "password": req.password, "track": req.track}
50    return {"id": user_id, "username": req.username}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    for u in users.values():
55        if u["username"] == req.username and u["password"] == req.password:
56            token = secrets.token_hex(16)
57            tokens[token] = u["id"]
58            return {"token": token}
59    raise HTTPException(status_code=401, detail="Invalid credentials")
60 
61@app.post("/proposals")
62def create_proposal(req: ProposalCreate, token: str = Header(...)):
63    global next_proposal_id
64    user_id = get_current_user(token)
65    pid = next_proposal_id
66    next_proposal_id += 1
67    proposals[pid] = {"id": pid, "title": req.title, "abstract": req.abstract, "track": req.track, "speaker": req.speaker, "user_id": user_id}
68    return proposals[pid]
69 
70@app.get("/proposals/{proposal_id}")
71def get_proposal(proposal_id: int, token: str = Header(...)):
72    get_current_user(token)
73    p = proposals.get(proposal_id)
74    if not p:
75        raise HTTPException(status_code=404, detail="Proposal not found")
76    return p
77 
78@app.post("/votes")
79def create_vote(req: VoteCreate, token: str = Header(...)):
80    global next_vote_id
81    user_id = get_current_user(token)
82    if req.proposal_id not in proposals:
83        raise HTTPException(status_code=404, detail="Proposal not found")
84    if req.score < 1 or req.score > 10:
85        raise HTTPException(status_code=400, detail="Score must be between 1 and 10")
86    vid = next_vote_id
87    next_vote_id += 1
88    votes[vid] = {"id": vid, "proposal_id": req.proposal_id, "score": req.score, "user_id": user_id}
89    return votes[vid]
90 
91@app.get("/votes/{vote_id}")
92def get_vote(vote_id: int, token: str = Header(...)):
93    get_current_user(token)
94    v = votes.get(vote_id)
95    if not v:
96        raise HTTPException(status_code=404, detail="Vote not found")
97    return v
98 
99@app.get("/tally/{track}")
100def tally_track(track: str, token: str = Header(...)):
101    get_current_user(token)
102    track_proposals = [p for p in proposals.values() if p["track"] == track]
103    result = {}
104    for p in track_proposals:
105        total = 0
106        count = 0
107        for v in votes.values():
108            if v["proposal_id"] == p["id"]:
109                total += v["score"]
110                count += 1
111        avg = total / count if count > 0 else 0
112        result[p["id"]] = {"title": p["title"], "total_score": total, "vote_count": count, "average": avg}
113    return result

1fastapi
2uvicorn