Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import uuid
5import hashlib
6import secrets
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12sessions = {}
13session_id_counter = 1
14 
15class SignupRequest(BaseModel):
16    email: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    email: str
21    password: str
22 
23class SessionCreate(BaseModel):
24    name: str
25    time: str
26    style: str
27    level: str = "beginner"
28    max_capacity: int = 20
29 
30class SessionUpdate(BaseModel):
31    name: Optional[str] = None
32    time: Optional[str] = None
33    style: Optional[str] = None
34    level: Optional[str] = None
35    max_capacity: Optional[int] = None
36 
37def get_current_user(authorization: str = Header(None)):
38    if not authorization:
39        raise HTTPException(status_code=401, detail="Missing auth header")
40    scheme, _, token = authorization.partition(" ")
41    if scheme.lower() != "bearer" or not token:
42        raise HTTPException(status_code=401, detail="Invalid auth header")
43    user_id = tokens.get(token)
44    if not user_id:
45        raise HTTPException(status_code=401, detail="Invalid token")
46    return user_id
47 
48@app.post("/signup")
49def signup(req: SignupRequest):
50    if req.email in users:
51        raise HTTPException(status_code=400, detail="User already exists")
52    user_id = str(uuid.uuid4())
53    users[req.email] = {"id": user_id, "email": req.email, "password": hashlib.sha256(req.password.encode()).hexdigest()}
54    token = secrets.token_hex(32)
55    tokens[token] = user_id
56    return {"token": token, "user_id": user_id}
57 
58@app.post("/login")
59def login(req: LoginRequest):
60    user = users.get(req.email)
61    if not user or user["password"] != hashlib.sha256(req.password.encode()).hexdigest():
62        raise HTTPException(status_code=401, detail="Invalid credentials")
63    token = secrets.token_hex(32)
64    tokens[token] = user["id"]
65    return {"token": token, "user_id": user["id"]}
66 
67@app.get("/sessions/{session_id}")
68def get_session(session_id: int, authorization: str = Header(None)):
69    get_current_user(authorization)
70    session = sessions.get(session_id)
71    if not session:
72        raise HTTPException(status_code=404, detail="Session not found")
73    return session
74 
75@app.post("/sessions")
76def create_session(req: SessionCreate, authorization: str = Header(None)):
77    get_current_user(authorization)
78    global session_id_counter
79    session_id = session_id_counter
80    session_id_counter += 1
81    sessions[session_id] = {
82        "id": session_id,
83        "name": req.name,
84        "time": req.time,
85        "style": req.style,
86        "level": req.level,
87        "max_capacity": req.max_capacity
88    }
89    return sessions[session_id]
90 
91@app.patch("/sessions/{session_id}")
92def update_session(session_id: int, req: SessionUpdate, authorization: str = Header(None)):
93    get_current_user(authorization)
94    session = sessions.get(session_id)
95    if not session:
96        raise HTTPException(status_code=404, detail="Session not found")
97    if req.name is not None:
98        session["name"] = req.name
99    if req.time is not None:
100        session["time"] = req.time
101    if req.style is not None:
102        session["style"] = req.style
103    if req.level is not None:
104        session["level"] = req.level
105    if req.max_capacity is not None:
106        session["max_capacity"] = req.max_capacity
107    return session

1fastapi
2uvicorn