Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import datetime
6import os
7import shutil
8from fastapi import UploadFile, File, Form
9 
10app = FastAPI()
11 
12users = {}
13groups = {}
14sessions = {}
15resources = {}
16files = {}
17next_user_id = 1
18next_group_id = 1
19next_session_id = 1
20next_resource_id = 1
21next_file_id = 1
22 
23def get_user_from_token(authorization: str = Header(None)):
24    if not authorization:
25        raise HTTPException(status_code=401, detail="No auth")
26    token = authorization.replace("Bearer ", "")
27    for uid, u in users.items():
28        if u.get("token") == token:
29            return uid, u
30    raise HTTPException(status_code=401, detail="Invalid token")
31 
32class SignupRequest(BaseModel):
33    username: str
34    password: str
35 
36class LoginRequest(BaseModel):
37    username: str
38    password: str
39 
40class CreateGroupRequest(BaseModel):
41    name: str
42 
43class PromoteRequest(BaseModel):
44    user_id: int
45 
46class CreateSessionRequest(BaseModel):
47    group_id: int
48    start_time: str
49    end_time: str
50 
51class CreateResourceRequest(BaseModel):
52    group_id: int
53    title: str
54    content: str = ""
55 
56@app.post("/signup")
57def signup(req: SignupRequest):
58    global next_user_id
59    for u in users.values():
60        if u["username"] == req.username:
61            raise HTTPException(status_code=400, detail="Username taken")
62    uid = next_user_id
63    next_user_id += 1
64    token = secrets.token_hex(16)
65    users[uid] = {
66        "id": uid,
67        "username": req.username,
68        "password": req.password,
69        "token": token,
70        "role": "student"
71    }
72    return {"user_id": uid, "token": token}
73 
74@app.post("/login")
75def login(req: LoginRequest):
76    for uid, u in users.items():
77        if u["username"] == req.username and u["password"] == req.password:
78            token = secrets.token_hex(16)
79            u["token"] = token
80            return {"user_id": uid, "token": token}
81    raise HTTPException(status_code=401, detail="Invalid credentials")
82 
83@app.post("/groups")
84def create_group(req: CreateGroupRequest, authorization: str = Header(None)):
85    global next_group_id
86    uid, u = get_user_from_token(authorization)
87    gid = next_group_id
88    next_group_id += 1
89    groups[gid] = {
90        "id": gid,
91        "name": req.name,
92        "leader_id": uid,
93        "tutors": [uid],
94        "members": [uid]
95    }
96    return {"group_id": gid}
97 
98@app.post("/group/{group_id}/promote")
99def promote_to_tutor(group_id: int, req: PromoteRequest, authorization: str = Header(None)):
100    uid, u = get_user_from_token(authorization)
101    g = groups.get(group_id)
102    if not g:
103        raise HTTPException(status_code=404, detail="Group not found")
104    if g["leader_id"] != uid:
105        raise HTTPException(status_code=403, detail="Only leader can promote")
106    if req.user_id not in g["members"]:
107        raise HTTPException(status_code=400, detail="User not in group")
108    if req.user_id not in g["tutors"]:
109        g["tutors"].append(req.user_id)
110    return {"status": "promoted", "user_id": req.user_id}
111 
112@app.post("/group/{group_id}/join")
113def join_group(group_id: int, authorization: str = Header(None)):
114    uid, u = get_user_from_token(authorization)
115    g = groups.get(group_id)
116    if not g:
117        raise HTTPException(status_code=404, detail="Group not found")
118    if uid not in g["members"]:
119        g["members"].append(uid)
120    return {"status": "joined", "group_id": group_id}
121 
122@app.post("/sessions")
123def create_session(req: CreateSessionRequest, authorization: str = Header(None)):
124    global next_session_id
125    uid, u = get_user_from_token(authorization)
126    g = groups.get(req.group_id)
127    if not g:
128        raise HTTPException(status_code=404, detail="Group not found")
129    if uid not in g["tutors"]:
130        raise HTTPException(status_code=403, detail="Only tutors can schedule sessions")
131    sid = next_session_id
132    next_session_id += 1
133    sessions[sid] = {
134        "id": sid,
135        "group_id": req.group_id,
136        "start_time": req.start_time,
137        "end_time": req.end_time,
138        "tutor_id": uid
139    }
140    return {"session_id": sid}
141 
142@app.get("/sessions/{session_id}")
143def get_session(session_id: int, authorization: str = Header(None)):
144    uid, u = get_user_from_token(authorization)
145    s = sessions.get(session_id)
146    if not s:
147        raise HTTPException(status_code=404, detail="Session not found")
148    return s
149 
150@app.post("/resources")
151def create_resource(req: CreateResourceRequest, authorization: str = Header(None)):
152    global next_resource_id
153    uid, u = get_user_from_token(authorization)
154    g = groups.get(req.group_id)
155    if not g:
156        raise HTTPException(status_code=404, detail="Group not found")
157    if uid not in g["members"]:
158        raise HTTPException(status_code=403, detail="Not a group member")
159    rid = next_resource_id
160    next_resource_id += 1
161    resources[rid] = {
162        "id": rid,
163        "group_id": req.group_id,
164        "title": req.title,
165        "content": req.content,
166        "uploader_id": uid
167    }
168    return {"resource_id": rid}
169 
170@app.get("/resources/{resource_id}")
171def get_resource(resource_id: int, authorization: str = Header(None)):
172    uid, u = get_user_from_token(authorization)
173    r = resources.get(resource_id)
174    if not r:
175        raise HTTPException(status_code=404, detail="Resource not found")
176    return r
177 
178UPLOAD_DIR = "uploads"
179os.makedirs(UPLOAD_DIR, exist_ok=True)
180 
181@app.post("/files/upload")
182async def upload_file(group_id: int = Form(...), file: UploadFile = File(...), authorization: str = Header(None)):
183    global next_file_id
184    uid, u = get_user_from_token(authorization)
185    g = groups.get(group_id)
186    if not g:
187        raise HTTPException(status_code=404, detail="Group not found")
188    if uid not in g["members"]:
189        raise HTTPException(status_code=403, detail="Not a group member")
190    fid = next_file_id
191    next_file_id += 1
192    file_path = os.path.join(UPLOAD_DIR, f"{fid}_{file.filename}")
193    with open(file_path, "wb") as f:
194        f.write(await file.read())
195    files[fid] = {
196        "id": fid,
197        "group_id": group_id,
198        "filename": file.filename,
199        "path": file_path,
200        "uploader_id": uid
201    }
202    return {"file_id": fid, "filename": file.filename}
203 
204@app.get("/files/{file_id}")
205def get_file(file_id: int, authorization: str = Header(None)):
206    uid, u = get_user_from_token(authorization)
207    f = files.get(file_id)
208    if not f:
209        raise HTTPException(status_code=404, detail="File not found")
210    return {"file_id": f["id"], "filename": f["filename"], "group_id": f["group_id"]}
211 
212@app.get("/users/{user_id}")
213def get_user(user_id: int, authorization: str = Header(None)):
214    uid, u = get_user_from_token(authorization)
215    user = users.get(user_id)
216    if not user:
217        raise HTTPException(status_code=404, detail="User not found")
218    return {"id": user["id"], "username": user["username"], "role": user["role"]}
219 
220@app.get("/groups/{group_id}")
221def get_group(group_id: int, authorization: str = Header(None)):
222    uid, u = get_user_from_token(authorization)
223    g = groups.get(group_id)
224    if not g:
225        raise HTTPException(status_code=404, detail="Group not found")
226    return g

1fastapi
2uvicorn
3python-multipart