Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List, Union
4import secrets
5import time
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11surveys = {}
12responses = {}
13next_user_id = 1
14next_survey_id = 1
15next_response_id = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class Question(BaseModel):
26    question_text: str
27    question_type: str  # "text" or "multiple_choice"
28    options: Optional[List[str]] = None
29 
30class SurveyCreate(BaseModel):
31    title: str
32    questions: List[Question]
33 
34class SurveyResponse(BaseModel):
35    answers: List[Union[str, List[str]]]
36 
37def get_user_from_token(authorization: str = Header(...)):
38    if not authorization.startswith("Bearer "):
39        raise HTTPException(status_code=401, detail="Invalid token")
40    token = authorization[7:]
41    user_id = tokens.get(token)
42    if user_id is None:
43        raise HTTPException(status_code=401, detail="Invalid token")
44    return user_id
45 
46@app.post("/signup")
47def signup(req: SignupRequest):
48    global next_user_id
49    for u in users.values():
50        if u["username"] == req.username:
51            raise HTTPException(status_code=400, detail="Username already exists")
52    user_id = next_user_id
53    next_user_id += 1
54    users[user_id] = {"username": req.username, "password": req.password}
55    return {"user_id": user_id, "message": "User created"}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for uid, u in users.items():
60        if u["username"] == req.username and u["password"] == req.password:
61            token = secrets.token_hex(32)
62            tokens[token] = uid
63            return {"token": token}
64    raise HTTPException(status_code=401, detail="Invalid credentials")
65 
66@app.post("/surveys")
67def create_survey(survey: SurveyCreate, authorization: str = Header(...)):
68    global next_survey_id
69    get_user_from_token(authorization)
70    survey_id = next_survey_id
71    next_survey_id += 1
72    surveys[survey_id] = {
73        "id": survey_id,
74        "title": survey.title,
75        "questions": [q.dict() for q in survey.questions]
76    }
77    responses[survey_id] = []
78    return {"survey_id": survey_id}
79 
80@app.get("/surveys/{survey_id}")
81def get_survey(survey_id: int, authorization: str = Header(...)):
82    get_user_from_token(authorization)
83    survey = surveys.get(survey_id)
84    if not survey:
85        raise HTTPException(status_code=404, detail="Survey not found")
86    return survey
87 
88@app.post("/surveys/{survey_id}/responses")
89def submit_response(survey_id: int, response: SurveyResponse, authorization: str = Header(...)):
90    get_user_from_token(authorization)
91    survey = surveys.get(survey_id)
92    if not survey:
93        raise HTTPException(status_code=404, detail="Survey not found")
94    if len(response.answers) != len(survey["questions"]):
95        raise HTTPException(status_code=400, detail="Answer count mismatch")
96    global next_response_id
97    response_id = next_response_id
98    next_response_id += 1
99    responses[survey_id].append({
100        "response_id": response_id,
101        "answers": response.answers,
102        "timestamp": time.time()
103    })
104    return {"response_id": response_id}
105 
106@app.get("/surveys/{survey_id}/responses/count")
107def get_response_count(survey_id: int, authorization: str = Header(...)):
108    get_user_from_token(authorization)
109    survey = surveys.get(survey_id)
110    if not survey:
111        raise HTTPException(status_code=404, detail="Survey not found")
112    return {"survey_id": survey_id, "response_count": len(responses.get(survey_id, []))}

1fastapi
2uvicorn