Browse the data

1from fastapi import FastAPI, Header, HTTPException
2from pydantic import BaseModel
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9polls = {}
10votes = {}
11 
12user_counter = 0
13poll_counter = 0
14vote_counter = 0
15 
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26 
27def get_user_from_token(authorization: str):
28    if not authorization:
29        raise HTTPException(status_code=401, detail="Missing authorization header")
30    token = authorization.replace("Bearer ", "").strip()
31    user_id = tokens.get(token)
32    if user_id is None:
33        raise HTTPException(status_code=401, detail="Invalid token")
34    return user_id
35 
36 
37@app.post("/signup")
38def signup(req: dict):
39    global user_counter
40    username = req.get("username")
41    password = req.get("password")
42    if not username or not password:
43        raise HTTPException(status_code=400, detail="username and password required")
44    for u in users.values():
45        if u["username"] == username:
46            raise HTTPException(status_code=400, detail="username taken")
47    user_counter += 1
48    user = {"id": user_counter, "username": username, "password": password}
49    for k, v in req.items():
50        if k not in ("username", "password"):
51            user[k] = v
52    users[user_counter] = user
53    return user
54 
55 
56@app.post("/login")
57def login(req: LoginRequest):
58    for user in users.values():
59        if user["username"] == req.username and user["password"] == req.password:
60            token = secrets.token_hex(16)
61            tokens[token] = user["id"]
62            return {"token": token}
63    raise HTTPException(status_code=401, detail="invalid credentials")
64 
65 
66@app.post("/polls")
67def create_poll(req: dict, authorization: str = Header(None)):
68    global poll_counter
69    user_id = get_user_from_token(authorization)
70    question = req.get("question")
71    options = req.get("options")
72    if not question or not options:
73        raise HTTPException(status_code=400, detail="question and options required")
74    poll_counter += 1
75    poll = {
76        "id": poll_counter,
77        "user_id": user_id,
78        "question": question,
79        "options": options,
80        "results": {opt: 0 for opt in options},
81    }
82    for k, v in req.items():
83        if k not in ("question", "options"):
84            poll[k] = v
85    polls[poll_counter] = poll
86    return poll
87 
88 
89@app.get("/polls/{poll_id}")
90def get_poll(poll_id: int):
91    poll = polls.get(poll_id)
92    if poll is None:
93        raise HTTPException(status_code=404, detail="poll not found")
94    return poll
95 
96 
97@app.post("/votes")
98def create_vote(req: dict, authorization: str = Header(None)):
99    global vote_counter
100    user_id = get_user_from_token(authorization)
101    poll_id = req.get("poll_id")
102    option = req.get("option")
103    if poll_id is None or option is None:
104        raise HTTPException(status_code=400, detail="poll_id and option required")
105    poll = polls.get(poll_id)
106    if poll is None:
107        raise HTTPException(status_code=404, detail="poll not found")
108    if option not in poll["options"]:
109        raise HTTPException(status_code=400, detail="invalid option")
110    vote_counter += 1
111    vote = {
112        "id": vote_counter,
113        "user_id": user_id,
114        "poll_id": poll_id,
115        "option": option,
116    }
117    for k, v in req.items():
118        if k not in ("poll_id", "option"):
119            vote[k] = v
120    votes[vote_counter] = vote
121    poll["results"][option] += 1
122    return vote
123 
124 
125@app.get("/votes/{vote_id}")
126def get_vote(vote_id: int):
127    vote = votes.get(vote_id)
128    if vote is None:
129        raise HTTPException(status_code=404, detail="vote not found")
130    return vote
131 
132 
133@app.get("/polls/{poll_id}/results")
134def get_results(poll_id: int):
135    poll = polls.get(poll_id)
136    if poll is None:
137        raise HTTPException(status_code=404, detail="poll not found")
138    return {"poll_id": poll_id, "question": poll["question"], "results": poll["results"]}

1fastapi
2uvicorn
3pydantic