Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import secrets
5import time
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11groups = {}
12user_groups = {}
13group_admins = {}
14group_members = {}
15next_user_id = 1
16next_group_id = 1
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class GroupCreate(BaseModel):
27    name: str
28    description: str = ""
29    privacy: str = "public"
30    invite_link_expiry: Optional[int] = None
31 
32class GroupUpdate(BaseModel):
33    name: Optional[str] = None
34    description: Optional[str] = None
35    privacy: Optional[str] = None
36    invite_link_expiry: Optional[int] = None
37    admin_privileges: Optional[dict] = None
38 
39def get_user_id_from_token(authorization: str = Header(...)):
40    if not authorization.startswith("Bearer "):
41        raise HTTPException(status_code=401, detail="Invalid token")
42    token = authorization[7:]
43    user_id = tokens.get(token)
44    if user_id is None:
45        raise HTTPException(status_code=401, detail="Invalid token")
46    return user_id
47 
48@app.post("/signup")
49def signup(req: SignupRequest):
50    global next_user_id
51    user_id = next_user_id
52    next_user_id += 1
53    users[user_id] = {"username": req.username, "password": req.password}
54    return {"user_id": user_id}
55 
56@app.post("/login")
57def login(req: LoginRequest):
58    for uid, u in users.items():
59        if u["username"] == req.username and u["password"] == req.password:
60            token = secrets.token_hex(16)
61            tokens[token] = uid
62            return {"token": token}
63    raise HTTPException(status_code=401, detail="Invalid credentials")
64 
65@app.get("/groups/{group_id}")
66def get_group(group_id: int):
67    group = groups.get(group_id)
68    if not group:
69        raise HTTPException(status_code=404, detail="Group not found")
70    return group
71 
72@app.post("/groups")
73def create_group(req: GroupCreate, authorization: str = Header(...)):
74    global next_group_id
75    user_id = get_user_id_from_token(authorization)
76    group_id = next_group_id
77    next_group_id += 1
78    groups[group_id] = {
79        "id": group_id,
80        "name": req.name,
81        "description": req.description,
82        "privacy": req.privacy,
83        "invite_link_expiry": req.invite_link_expiry,
84        "created_by": user_id,
85        "created_at": int(time.time())
86    }
87    group_admins[group_id] = [user_id]
88    group_members[group_id] = [user_id]
89    if user_id not in user_groups:
90        user_groups[user_id] = []
91    user_groups[user_id].append(group_id)
92    return groups[group_id]
93 
94@app.patch("/groups/{group_id}")
95def update_group(group_id: int, req: GroupUpdate, authorization: str = Header(...)):
96    user_id = get_user_id_from_token(authorization)
97    group = groups.get(group_id)
98    if not group:
99        raise HTTPException(status_code=404, detail="Group not found")
100    if user_id not in group_admins.get(group_id, []):
101        raise HTTPException(status_code=403, detail="Not an admin")
102    if req.name is not None:
103        group["name"] = req.name
104    if req.description is not None:
105        group["description"] = req.description
106    if req.privacy is not None:
107        group["privacy"] = req.privacy
108    if req.invite_link_expiry is not None:
109        group["invite_link_expiry"] = req.invite_link_expiry
110    if req.admin_privileges is not None:
111        group["admin_privileges"] = req.admin_privileges
112    return group

1fastapi
2uvicorn