Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import datetime
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11challenges = {}
12challenge_id_counter = 1
13leaderboard = {}
14daily_progress = {}
15 
16def get_current_user(authorization: Optional[str] = Header(None)):
17    if not authorization:
18        raise HTTPException(status_code=401, detail="Missing auth header")
19    token = authorization.replace("Bearer ", "")
20    user_id = tokens.get(token)
21    if not user_id:
22        raise HTTPException(status_code=401, detail="Invalid token")
23    return user_id
24 
25class SignupRequest(BaseModel):
26    username: str
27    password: str
28 
29class LoginRequest(BaseModel):
30    username: str
31    password: str
32 
33class ChallengeCreate(BaseModel):
34    name: str
35    goal: str
36    duration: int
37    participants: list[str]
38    reward: str
39 
40class ChallengeUpdate(BaseModel):
41    name: Optional[str] = None
42    goal: Optional[str] = None
43    duration: Optional[int] = None
44    participants: Optional[list[str]] = None
45    reward: Optional[str] = None
46 
47class ProgressUpdate(BaseModel):
48    user: str
49    progress: int
50 
51@app.post("/signup")
52def signup(req: SignupRequest):
53    if req.username in users:
54        raise HTTPException(status_code=400, detail="User exists")
55    users[req.username] = req.password
56    token = secrets.token_hex(16)
57    tokens[token] = req.username
58    return {"token": token}
59 
60@app.post("/login")
61def login(req: LoginRequest):
62    if users.get(req.username) != req.password:
63        raise HTTPException(status_code=401, detail="Invalid credentials")
64    token = secrets.token_hex(16)
65    tokens[token] = req.username
66    return {"token": token}
67 
68@app.get("/challenges/{challenge_id}")
69def get_challenge(challenge_id: int, authorization: Optional[str] = Header(None)):
70    get_current_user(authorization)
71    challenge = challenges.get(challenge_id)
72    if not challenge:
73        raise HTTPException(status_code=404, detail="Challenge not found")
74    return challenge
75 
76@app.post("/challenges")
77def create_challenge(challenge: ChallengeCreate, authorization: Optional[str] = Header(None)):
78    user = get_current_user(authorization)
79    global challenge_id_counter
80    cid = challenge_id_counter
81    challenge_id_counter += 1
82    challenges[cid] = {
83        "id": cid,
84        "name": challenge.name,
85        "goal": challenge.goal,
86        "duration": challenge.duration,
87        "participants": challenge.participants,
88        "reward": challenge.reward,
89        "created_by": user
90    }
91    leaderboard[cid] = {p: 0 for p in challenge.participants}
92    daily_progress[cid] = {}
93    return challenges[cid]
94 
95@app.patch("/challenges/{challenge_id}")
96def update_challenge(challenge_id: int, update: ChallengeUpdate, authorization: Optional[str] = Header(None)):
97    user = get_current_user(authorization)
98    challenge = challenges.get(challenge_id)
99    if not challenge:
100        raise HTTPException(status_code=404, detail="Challenge not found")
101    if challenge["created_by"] != user:
102        raise HTTPException(status_code=403, detail="Not your challenge")
103    if update.name is not None:
104        challenge["name"] = update.name
105    if update.goal is not None:
106        challenge["goal"] = update.goal
107    if update.duration is not None:
108        challenge["duration"] = update.duration
109    if update.participants is not None:
110        challenge["participants"] = update.participants
111        if challenge_id not in leaderboard:
112            leaderboard[challenge_id] = {}
113        for p in update.participants:
114            if p not in leaderboard[challenge_id]:
115                leaderboard[challenge_id][p] = 0
116    if update.reward is not None:
117        challenge["reward"] = update.reward
118    return challenge
119 
120@app.post("/challenges/{challenge_id}/progress")
121def add_progress(challenge_id: int, progress: ProgressUpdate, authorization: Optional[str] = Header(None)):
122    get_current_user(authorization)
123    challenge = challenges.get(challenge_id)
124    if not challenge:
125        raise HTTPException(status_code=404, detail="Challenge not found")
126    if progress.user not in challenge["participants"]:
127        raise HTTPException(status_code=400, detail="User not in challenge")
128    today = str(datetime.date.today())
129    if challenge_id not in daily_progress:
130        daily_progress[challenge_id] = {}
131    if today not in daily_progress[challenge_id]:
132        daily_progress[challenge_id][today] = {}
133    daily_progress[challenge_id][today][progress.user] = daily_progress[challenge_id][today].get(progress.user, 0) + progress.progress
134    if challenge_id not in leaderboard:
135        leaderboard[challenge_id] = {}
136    leaderboard[challenge_id][progress.user] = leaderboard[challenge_id].get(progress.user, 0) + progress.progress
137    return {"message": "Progress added"}
138 
139@app.get("/challenges/{challenge_id}/leaderboard")
140def get_leaderboard(challenge_id: int, authorization: Optional[str] = Header(None)):
141    get_current_user(authorization)
142    challenge = challenges.get(challenge_id)
143    if not challenge:
144        raise HTTPException(status_code=404, detail="Challenge not found")
145    lb = leaderboard.get(challenge_id, {})
146    sorted_lb = sorted(lb.items(), key=lambda x: x[1], reverse=True)
147    return {"leaderboard": [{"user": u, "score": s} for u, s in sorted_lb]}
148 
149@app.get("/challenges/{challenge_id}/daily")
150def get_daily_progress(challenge_id: int, authorization: Optional[str] = Header(None)):
151    get_current_user(authorization)
152    challenge = challenges.get(challenge_id)
153    if not challenge:
154        raise HTTPException(status_code=404, detail="Challenge not found")
155    return daily_progress.get(challenge_id, {})

1fastapi
2uvicorn