Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4import uvicorn
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10subscriptions = {}
11orders = {}
12boxes = {}
13curators = set()
14next_user_id = 1
15next_sub_id = 1
16next_order_id = 1
17next_box_id = 1
18 
19def get_current_user(authorization: Optional[str] = Header(None)):
20    if not authorization:
21        raise HTTPException(status_code=401, detail="No auth header")
22    token = authorization.replace("Bearer ", "")
23    if token not in tokens:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return tokens[token]
26 
27@app.post("/signup")
28def signup(username: str, password: str):
29    global next_user_id
30    if any(u["username"] == username for u in users.values()):
31        raise HTTPException(status_code=400, detail="Username taken")
32    user_id = next_user_id
33    users[user_id] = {"id": user_id, "username": username, "password": password}
34    next_user_id += 1
35    if len(users) == 1:
36        curators.add(user_id)
37    token = secrets.token_hex(16)
38    tokens[token] = user_id
39    return {"user_id": user_id, "token": token}
40 
41@app.post("/login")
42def login(username: str, password: str):
43    for uid, u in users.items():
44        if u["username"] == username and u["password"] == password:
45            token = secrets.token_hex(16)
46            tokens[token] = uid
47            return {"token": token}
48    raise HTTPException(status_code=401, detail="Invalid credentials")
49 
50@app.post("/community/promote")
51def promote(user_id: int, authorization: Optional[str] = Header(None)):
52    current_user = get_current_user(authorization)
53    if current_user not in curators:
54        raise HTTPException(status_code=403, detail="Not a curator")
55    if user_id not in users:
56        raise HTTPException(status_code=404, detail="User not found")
57    curators.add(user_id)
58    return {"message": f"User {user_id} promoted to curator"}
59 
60@app.post("/subscription")
61def create_subscription(user_id: int, billing_cycle: str = "monthly", authorization: Optional[str] = Header(None)):
62    current_user = get_current_user(authorization)
63    if current_user not in curators:
64        raise HTTPException(status_code=403, detail="Not a curator")
65    global next_sub_id
66    sub_id = next_sub_id
67    subscriptions[sub_id] = {"id": sub_id, "user_id": user_id, "billing_cycle": billing_cycle}
68    next_sub_id += 1
69    return subscriptions[sub_id]
70 
71@app.get("/subscription/{sub_id}")
72def get_subscription(sub_id: int, authorization: Optional[str] = Header(None)):
73    get_current_user(authorization)
74    if sub_id not in subscriptions:
75        raise HTTPException(status_code=404, detail="Subscription not found")
76    return subscriptions[sub_id]
77 
78@app.post("/order")
79def create_order(subscription_id: int, box_customization: str = "default", authorization: Optional[str] = Header(None)):
80    current_user = get_current_user(authorization)
81    if subscription_id not in subscriptions:
82        raise HTTPException(status_code=404, detail="Subscription not found")
83    global next_order_id
84    order_id = next_order_id
85    orders[order_id] = {"id": order_id, "subscription_id": subscription_id, "box_customization": box_customization, "user_id": current_user}
86    next_order_id += 1
87    return orders[order_id]
88 
89@app.get("/order/{order_id}")
90def get_order(order_id: int, authorization: Optional[str] = Header(None)):
91    get_current_user(authorization)
92    if order_id not in orders:
93        raise HTTPException(status_code=404, detail="Order not found")
94    return orders[order_id]
95 
96@app.post("/box")
97def create_box(order_id: int, contents: str = "surprise", authorization: Optional[str] = Header(None)):
98    current_user = get_current_user(authorization)
99    if order_id not in orders:
100        raise HTTPException(status_code=404, detail="Order not found")
101    global next_box_id
102    box_id = next_box_id
103    boxes[box_id] = {"id": box_id, "order_id": order_id, "contents": contents, "user_id": current_user}
104    next_box_id += 1
105    return boxes[box_id]
106 
107@app.get("/box/{box_id}")
108def get_box(box_id: int, authorization: Optional[str] = Header(None)):
109    get_current_user(authorization)
110    if box_id not in boxes:
111        raise HTTPException(status_code=404, detail="Box not found")
112    return boxes[box_id]

1fastapi
2uvicorn