Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import random
5import string
6from datetime import datetime, timedelta
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12polls = {}
13poll_id_counter = 1
14 
15def generate_token():
16    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
17 
18def get_current_user(authorization: str = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing auth token")
21    token = authorization.replace("Bearer ", "")
22    if token not in tokens:
23        raise HTTPException(status_code=401, detail="Invalid token")
24    return tokens[token]
25 
26class SignupRequest(BaseModel):
27    username: str
28    password: str
29 
30class LoginRequest(BaseModel):
31    username: str
32    password: str
33 
34class PollCreate(BaseModel):
35    question: str
36    options: List[str]
37    expiration: Optional[str] = None
38    visibility: str = "public"
39    allow_multiple_votes: bool = False
40    results_privacy: str = "public"
41 
42class PollUpdate(BaseModel):
43    question: Optional[str] = None
44    options: Optional[List[str]] = None
45    expiration: Optional[str] = None
46    visibility: Optional[str] = None
47    allow_multiple_votes: Optional[bool] = None
48    results_privacy: Optional[str] = None
49 
50class VoteRequest(BaseModel):
51    option_index: int
52 
53@app.post("/signup")
54def signup(req: SignupRequest):
55    if req.username in users:
56        raise HTTPException(status_code=400, detail="User exists")
57    users[req.username] = {"password": req.password, "votes": {}}
58    token = generate_token()
59    tokens[token] = req.username
60    return {"token": token}
61 
62@app.post("/login")
63def login(req: LoginRequest):
64    if req.username not in users or users[req.username]["password"] != req.password:
65        raise HTTPException(status_code=401, detail="Invalid credentials")
66    token = generate_token()
67    tokens[token] = req.username
68    return {"token": token}
69 
70@app.post("/polls")
71def create_poll(poll: PollCreate, authorization: str = Header(None)):
72    user = get_current_user(authorization)
73    global poll_id_counter
74    poll_id = poll_id_counter
75    poll_id_counter += 1
76    polls[poll_id] = {
77        "id": poll_id,
78        "question": poll.question,
79        "options": poll.options,
80        "expiration": poll.expiration,
81        "visibility": poll.visibility,
82        "allow_multiple_votes": poll.allow_multiple_votes,
83        "results_privacy": poll.results_privacy,
84        "creator": user,
85        "votes": {i: [] for i in range(len(poll.options))}
86    }
87    return polls[poll_id]
88 
89@app.get("/polls/{poll_id}")
90def get_poll(poll_id: int, authorization: str = Header(None)):
91    user = get_current_user(authorization)
92    if poll_id not in polls:
93        raise HTTPException(status_code=404, detail="Poll not found")
94    poll = polls[poll_id]
95    if poll["results_privacy"] == "private" and poll["creator"] != user:
96        return {
97            "id": poll["id"],
98            "question": poll["question"],
99            "options": poll["options"],
100            "expiration": poll["expiration"],
101            "visibility": poll["visibility"],
102            "allow_multiple_votes": poll["allow_multiple_votes"],
103            "results_privacy": poll["results_privacy"],
104            "creator": poll["creator"],
105            "votes": {i: len(v) for i, v in poll["votes"].items()}
106        }
107    return poll
108 
109@app.put("/polls/{poll_id}")
110def update_poll(poll_id: int, update: PollUpdate, authorization: str = Header(None)):
111    user = get_current_user(authorization)
112    if poll_id not in polls:
113        raise HTTPException(status_code=404, detail="Poll not found")
114    poll = polls[poll_id]
115    if poll["creator"] != user:
116        raise HTTPException(status_code=403, detail="Not your poll")
117    if update.question is not None:
118        poll["question"] = update.question
119    if update.options is not None:
120        poll["options"] = update.options
121        poll["votes"] = {i: [] for i in range(len(update.options))}
122    if update.expiration is not None:
123        poll["expiration"] = update.expiration
124    if update.visibility is not None:
125        poll["visibility"] = update.visibility
126    if update.allow_multiple_votes is not None:
127        poll["allow_multiple_votes"] = update.allow_multiple_votes
128    if update.results_privacy is not None:
129        poll["results_privacy"] = update.results_privacy
130    return poll
131 
132@app.post("/polls/{poll_id}/vote")
133def vote(poll_id: int, vote: VoteRequest, authorization: str = Header(None)):
134    user = get_current_user(authorization)
135    if poll_id not in polls:
136        raise HTTPException(status_code=404, detail="Poll not found")
137    poll = polls[poll_id]
138    if vote.option_index < 0 or vote.option_index >= len(poll["options"]):
139        raise HTTPException(status_code=400, detail="Invalid option")
140    if not poll["allow_multiple_votes"]:
141        for option_votes in poll["votes"].values():
142            if user in option_votes:
143                raise HTTPException(status_code=400, detail="Already voted")
144    poll["votes"][vote.option_index].append(user)
145    return {"message": "Voted"}

1fastapi
2uvicorn
3pydantic