Browse the data

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

1fastapi>=0.104.0
2uvicorn>=0.24.0