Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10trainers = {}
11classes = {}
12enrollments = {}
13waitlist = {}
14next_user_id = 1
15next_trainer_id = 1
16next_class_id = 1
17next_enrollment_id = 1
18next_waitlist_id = 1
19 
20class UserSignup(BaseModel):
21    username: str
22    password: str
23 
24class UserLogin(BaseModel):
25    username: str
26    password: str
27 
28class TrainerCreate(BaseModel):
29    name: str
30 
31class ClassCreate(BaseModel):
32    trainer_id: int
33    max_capacity: int
34    difficulty: str
35 
36class JoinClass(BaseModel):
37    class_id: int
38    user_id: int
39 
40def get_current_user(authorization: Optional[str] = Header(None)):
41    if not authorization:
42        raise HTTPException(status_code=401, detail="Missing auth token")
43    token = authorization.replace("Bearer ", "")
44    for uid, t in tokens.items():
45        if t == token:
46            return uid
47    raise HTTPException(status_code=401, detail="Invalid token")
48 
49@app.post("/signup")
50def signup(body: UserSignup):
51    global next_user_id
52    uid = next_user_id
53    next_user_id += 1
54    users[uid] = {"id": uid, "username": body.username, "password": body.password}
55    return {"user_id": uid}
56 
57@app.post("/login")
58def login(body: UserLogin):
59    for uid, u in users.items():
60        if u["username"] == body.username and u["password"] == body.password:
61            token = secrets.token_hex(16)
62            tokens[uid] = token
63            return {"token": token}
64    raise HTTPException(status_code=401, detail="Invalid credentials")
65 
66@app.post("/trainers")
67def create_trainer(body: TrainerCreate, authorization: Optional[str] = Header(None)):
68    get_current_user(authorization)
69    global next_trainer_id
70    tid = next_trainer_id
71    next_trainer_id += 1
72    trainers[tid] = {"id": tid, "name": body.name}
73    return {"trainer_id": tid}
74 
75@app.get("/trainers/{trainer_id}")
76def get_trainer(trainer_id: int, authorization: Optional[str] = Header(None)):
77    get_current_user(authorization)
78    if trainer_id not in trainers:
79        raise HTTPException(status_code=404, detail="Trainer not found")
80    return trainers[trainer_id]
81 
82@app.post("/classes")
83def create_class(body: ClassCreate, authorization: Optional[str] = Header(None)):
84    get_current_user(authorization)
85    global next_class_id
86    if body.trainer_id not in trainers:
87        raise HTTPException(status_code=404, detail="Trainer not found")
88    cid = next_class_id
89    next_class_id += 1
90    classes[cid] = {
91        "id": cid,
92        "trainer_id": body.trainer_id,
93        "max_capacity": body.max_capacity,
94        "difficulty": body.difficulty,
95        "enrolled_count": 0
96    }
97    return {"class_id": cid}
98 
99@app.get("/classes/{class_id}")
100def get_class(class_id: int, authorization: Optional[str] = Header(None)):
101    get_current_user(authorization)
102    if class_id not in classes:
103        raise HTTPException(status_code=404, detail="Class not found")
104    return classes[class_id]
105 
106@app.post("/enroll")
107def enroll(body: JoinClass, authorization: Optional[str] = Header(None)):
108    get_current_user(authorization)
109    global next_enrollment_id, next_waitlist_id
110    c = classes.get(body.class_id)
111    if not c:
112        raise HTTPException(status_code=404, detail="Class not found")
113    if body.user_id not in users:
114        raise HTTPException(status_code=404, detail="User not found")
115    if c["enrolled_count"] < c["max_capacity"]:
116        eid = next_enrollment_id
117        next_enrollment_id += 1
118        enrollments[eid] = {"id": eid, "class_id": body.class_id, "user_id": body.user_id}
119        c["enrolled_count"] += 1
120        return {"enrollment_id": eid, "status": "enrolled"}
121    else:
122        wid = next_waitlist_id
123        next_waitlist_id += 1
124        waitlist[wid] = {"id": wid, "class_id": body.class_id, "user_id": body.user_id}
125        return {"waitlist_id": wid, "status": "waitlisted"}
126 
127@app.post("/drop")
128def drop(body: JoinClass, authorization: Optional[str] = Header(None)):
129    get_current_user(authorization)
130    c = classes.get(body.class_id)
131    if not c:
132        raise HTTPException(status_code=404, detail="Class not found")
133    for eid, e in list(enrollments.items()):
134        if e["class_id"] == body.class_id and e["user_id"] == body.user_id:
135            del enrollments[eid]
136            c["enrolled_count"] -= 1
137            # auto-promote from waitlist
138            waitlist_ids = sorted([wid for wid, w in waitlist.items() if w["class_id"] == body.class_id])
139            if waitlist_ids:
140                next_wid = waitlist_ids[0]
141                next_user = waitlist[next_wid]["user_id"]
142                del waitlist[next_wid]
143                eid_new = next_enrollment_id
144                next_enrollment_id += 1
145                enrollments[eid_new] = {"id": eid_new, "class_id": body.class_id, "user_id": next_user}
146                c["enrolled_count"] += 1
147                return {"status": "dropped", "promoted_user_id": next_user}
148            return {"status": "dropped"}
149    raise HTTPException(status_code=404, detail="Enrollment not found")
150 
151@app.get("/users/{user_id}")
152def get_user(user_id: int, authorization: Optional[str] = Header(None)):
153    get_current_user(authorization)
154    if user_id not in users:
155        raise HTTPException(status_code=404, detail="User not found")
156    return users[user_id]

1fastapi
2uvicorn