Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9videos = {}
10promotions = {}
11id_counter = {"users": 0, "videos": 0, "promotions": 0}
12partners = set()
13 
14def get_current_user(authorization: Optional[str] = Header(None)):
15    if not authorization:
16        raise HTTPException(status_code=401, detail="Missing auth token")
17    token = authorization.replace("Bearer ", "")
18    user_id = tokens.get(token)
19    if user_id is None:
20        raise HTTPException(status_code=401, detail="Invalid token")
21    return user_id
22 
23@app.post("/signup")
24def signup(username: str, password: str):
25    global id_counter
26    id_counter["users"] += 1
27    user_id = id_counter["users"]
28    users[user_id] = {"id": user_id, "username": username, "password": password, "partner": False}
29    return {"user_id": user_id, "username": username}
30 
31@app.post("/login")
32def login(username: str, password: str):
33    for uid, u in users.items():
34        if u["username"] == username and u["password"] == password:
35            token = secrets.token_hex(16)
36            tokens[token] = uid
37            return {"token": token}
38    raise HTTPException(status_code=401, detail="Invalid credentials")
39 
40@app.post("/promote/{user_id}")
41def promote_user(user_id: int, authorization: Optional[str] = Header(None)):
42    admin_id = get_current_user(authorization)
43    admin = users.get(admin_id)
44    if not admin or not admin.get("admin"):
45        raise HTTPException(status_code=403, detail="Only admins can promote")
46    target = users.get(user_id)
47    if not target:
48        raise HTTPException(status_code=404, detail="User not found")
49    target["partner"] = True
50    partners.add(user_id)
51    return {"message": f"User {user_id} promoted to partner"}
52 
53@app.get("/{resource}/{resource_id}")
54def get_resource(resource: str, resource_id: int):
55    if resource == "users":
56        item = users.get(resource_id)
57    elif resource == "videos":
58        item = videos.get(resource_id)
59    elif resource == "promotions":
60        item = promotions.get(resource_id)
61    else:
62        raise HTTPException(status_code=404, detail="Resource not found")
63    if not item:
64        raise HTTPException(status_code=404, detail="Not found")
65    return item
66 
67@app.post("/{resource}")
68def create_resource(resource: str, name: str, authorization: Optional[str] = Header(None)):
69    global id_counter
70    user_id = get_current_user(authorization)
71    if resource == "videos":
72        id_counter["videos"] += 1
73        vid = id_counter["videos"]
74        videos[vid] = {"id": vid, "name": name, "user_id": user_id}
75        return videos[vid]
76    elif resource == "promotions":
77        id_counter["promotions"] += 1
78        pid = id_counter["promotions"]
79        promotions[pid] = {"id": pid, "name": name, "user_id": user_id}
80        return promotions[pid]
81    else:
82        raise HTTPException(status_code=400, detail="Unsupported resource")

1fastapi
2uvicorn