Browse the data

1import csv
2import io
3import os
4from fastapi import FastAPI, Header, HTTPException, status
5from fastapi.responses import StreamingResponse
6 
7app = FastAPI()
8 
9users = {}
10next_user_id = 1
11tokens = {}
12 
13surveys = {}
14next_survey_id = 1
15responses = {}
16next_response_id = 1
17 
18def get_current_user(authorization: str = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing authorization header")
21    token = authorization.replace("Bearer ", "")
22    user_id = tokens.get(token)
23    if not user_id:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return user_id
26 
27@app.post("/signup")
28def signup(username: str, password: str):
29    global next_user_id
30    user_id = next_user_id
31    next_user_id += 1
32    users[user_id] = {"username": username, "password": password}
33    return {"user_id": user_id}
34 
35@app.post("/login")
36def login(username: str, password: str):
37    for uid, u in users.items():
38        if u["username"] == username and u["password"] == password:
39            token = f"token_{uid}_{username}"
40            tokens[token] = uid
41            return {"token": token}
42    raise HTTPException(status_code=401, detail="Invalid credentials")
43 
44@app.post("/surveys")
45def create_survey(name: str, authorization: str = Header(None)):
46    user_id = get_current_user(authorization)
47    global next_survey_id
48    survey_id = next_survey_id
49    next_survey_id += 1
50    surveys[survey_id] = {"id": survey_id, "name": name, "owner_id": user_id}
51    return surveys[survey_id]
52 
53@app.get("/surveys/{survey_id}")
54def get_survey(survey_id: int, authorization: str = Header(None)):
55    get_current_user(authorization)
56    survey = surveys.get(survey_id)
57    if not survey:
58        raise HTTPException(status_code=404, detail="Survey not found")
59    return survey
60 
61@app.post("/surveys/{survey_id}/responses")
62def create_response(survey_id: int, answer: str, authorization: str = Header(None)):
63    get_current_user(authorization)
64    if survey_id not in surveys:
65        raise HTTPException(status_code=404, detail="Survey not found")
66    global next_response_id
67    response_id = next_response_id
68    next_response_id += 1
69    responses[response_id] = {"id": response_id, "survey_id": survey_id, "answer": answer}
70    return responses[response_id]
71 
72@app.get("/surveys/{survey_id}/responses/export")
73def export_responses(survey_id: int, format: str = "csv", authorization: str = Header(None)):
74    get_current_user(authorization)
75    if survey_id not in surveys:
76        raise HTTPException(status_code=404, detail="Survey not found")
77    
78    survey_responses = [r for r in responses.values() if r["survey_id"] == survey_id]
79    
80    output = io.StringIO()
81    writer = csv.writer(output)
82    writer.writerow(["id", "survey_id", "answer"])
83    for r in survey_responses:
84        writer.writerow([r["id"], r["survey_id"], r["answer"]])
85    
86    output.seek(0)
87    
88    survey_dir = f"survey_data/{survey_id}"
89    os.makedirs(survey_dir, exist_ok=True)
90    with open(f"{survey_dir}/responses.csv", "w") as f:
91        f.write(output.getvalue())
92    
93    output.seek(0)
94    return StreamingResponse(output, media_type="text/csv", headers={"Content-Disposition": f"attachment; filename=responses.csv"})

1fastapi
2uvicorn