Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import uuid
5 
6app = FastAPI()
7 
8users = {}
9groups = {}
10group_id_counter = 1
11user_id_counter = 1
12tokens = {}  # token -> user_id
13 
14class SignupRequest(BaseModel):
15    username: str
16    password: str
17 
18class LoginRequest(BaseModel):
19    username: str
20    password: str
21 
22class CreateGroupRequest(BaseModel):
23    name: str
24    language: str
25 
26class PromoteRequest(BaseModel):
27    user_id: int
28 
29class LessonNote(BaseModel):
30    content: str
31 
32class VoiceChatLog(BaseModel):
33    user_id: int
34    message: str
35 
36def get_current_user(authorization: Optional[str] = Header(None)):
37    if not authorization:
38        raise HTTPException(status_code=401, detail="Missing auth token")
39    token = authorization.replace("Bearer ", "")
40    user_id = tokens.get(token)
41    if not user_id:
42        raise HTTPException(status_code=401, detail="Invalid token")
43    return user_id
44 
45@app.post("/signup")
46def signup(req: SignupRequest):
47    global user_id_counter
48    for u in users.values():
49        if u["username"] == req.username:
50            raise HTTPException(status_code=400, detail="Username taken")
51    user_id = user_id_counter
52    user_id_counter += 1
53    users[user_id] = {
54        "id": user_id,
55        "username": req.username,
56        "password": req.password,
57        "groups": []
58    }
59    token = str(uuid.uuid4())
60    tokens[token] = user_id
61    return {"user_id": user_id, "token": token}
62 
63@app.post("/login")
64def login(req: LoginRequest):
65    for user_id, user in users.items():
66        if user["username"] == req.username and user["password"] == req.password:
67            token = str(uuid.uuid4())
68            tokens[token] = user_id
69            return {"user_id": user_id, "token": token}
70    raise HTTPException(status_code=401, detail="Invalid credentials")
71 
72@app.get("/user/{user_id}")
73def get_user(user_id: int):
74    user = users.get(user_id)
75    if not user:
76        raise HTTPException(status_code=404, detail="User not found")
77    return user
78 
79@app.get("/group/{group_id}")
80def get_group(group_id: int):
81    group = groups.get(group_id)
82    if not group:
83        raise HTTPException(status_code=404, detail="Group not found")
84    return group
85 
86@app.post("/group")
87def create_group(req: CreateGroupRequest, authorization: Optional[str] = Header(None)):
88    global group_id_counter
89    user_id = get_current_user(authorization)
90    group_id = group_id_counter
91    group_id_counter += 1
92    groups[group_id] = {
93        "id": group_id,
94        "name": req.name,
95        "language": req.language,
96        "moderator": user_id,
97        "members": [user_id],
98        "tutors": [],
99        "lesson_notes": [],
100        "voice_chat_logs": []
101    }
102    users[user_id]["groups"].append(group_id)
103    return groups[group_id]
104 
105@app.post("/group/{group_id}/promote")
106def promote_to_tutor(group_id: int, req: PromoteRequest, authorization: Optional[str] = Header(None)):
107    user_id = get_current_user(authorization)
108    group = groups.get(group_id)
109    if not group:
110        raise HTTPException(status_code=404, detail="Group not found")
111    if group["moderator"] != user_id:
112        raise HTTPException(status_code=403, detail="Only moderator can promote")
113    if req.user_id not in group["members"]:
114        raise HTTPException(status_code=400, detail="User is not a member")
115    if req.user_id in group["tutors"]:
116        raise HTTPException(status_code=400, detail="Already a tutor")
117    group["tutors"].append(req.user_id)
118    return {"status": "promoted", "user_id": req.user_id}
119 
120@app.post("/group/{group_id}/lesson_notes")
121def add_lesson_note(group_id: int, note: LessonNote, authorization: Optional[str] = Header(None)):
122    user_id = get_current_user(authorization)
123    group = groups.get(group_id)
124    if not group:
125        raise HTTPException(status_code=404, detail="Group not found")
126    if user_id not in group["members"]:
127        raise HTTPException(status_code=403, detail="Not a member")
128    group["lesson_notes"].append({"user_id": user_id, "content": note.content})
129    return {"status": "added"}
130 
131@app.get("/group/{group_id}/lesson_notes")
132def get_lesson_notes(group_id: int, authorization: Optional[str] = Header(None)):
133    user_id = get_current_user(authorization)
134    group = groups.get(group_id)
135    if not group:
136        raise HTTPException(status_code=404, detail="Group not found")
137    if user_id not in group["members"]:
138        raise HTTPException(status_code=403, detail="Not a member")
139    return group["lesson_notes"]
140 
141@app.post("/group/{group_id}/voice_chat_logs")
142def add_voice_chat_log(group_id: int, log: VoiceChatLog, authorization: Optional[str] = Header(None)):
143    user_id = get_current_user(authorization)
144    group = groups.get(group_id)
145    if not group:
146        raise HTTPException(status_code=404, detail="Group not found")
147    if user_id not in group["members"]:
148        raise HTTPException(status_code=403, detail="Not a member")
149    group["voice_chat_logs"].append({"user_id": log.user_id, "message": log.message})
150    return {"status": "added"}
151 
152@app.get("/group/{group_id}/voice_chat_logs")
153def get_voice_chat_logs(group_id: int, authorization: Optional[str] = Header(None)):
154    user_id = get_current_user(authorization)
155    group = groups.get(group_id)
156    if not group:
157        raise HTTPException(status_code=404, detail="Group not found")
158    if user_id not in group["members"]:
159        raise HTTPException(status_code=403, detail="Not a member")
160    return group["voice_chat_logs"]

1fastapi
2uvicorn