Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import random
5import string
6 
7app = FastAPI()
8 
9users = {}
10orders = {}
11rewards = {}
12next_user_id = 1
13next_order_id = 1
14next_reward_id = 1
15tokens = {}
16 
17def generate_token():
18    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
19 
20def get_current_user(authorization: Optional[str] = Header(None)):
21    if not authorization or not authorization.startswith("Bearer "):
22        raise HTTPException(status_code=401, detail="Invalid auth")
23    token = authorization.split(" ")[1]
24    if token not in tokens:
25        raise HTTPException(status_code=401, detail="Invalid token")
26    return tokens[token]
27 
28class SignupRequest(BaseModel):
29    phone: str
30    name: str
31 
32class LoginRequest(BaseModel):
33    phone: str
34 
35class OrderRequest(BaseModel):
36    user_id: int
37    amount: float
38 
39class RewardRequest(BaseModel):
40    name: str
41    points_required: int
42    description: str = ""
43 
44class MakeManagerRequest(BaseModel):
45    pass
46 
47@app.post("/signup")
48def signup(req: SignupRequest):
49    global next_user_id
50    user_id = next_user_id
51    next_user_id += 1
52    users[user_id] = {
53        "id": user_id,
54        "phone": req.phone,
55        "name": req.name,
56        "points": 0,
57        "is_manager": False
58    }
59    return {"user_id": user_id, "message": "User created"}
60 
61@app.post("/login")
62def login(req: LoginRequest):
63    for uid, u in users.items():
64        if u["phone"] == req.phone:
65            token = generate_token()
66            tokens[token] = uid
67            return {"token": token}
68    raise HTTPException(status_code=404, detail="User not found")
69 
70@app.get("/users/{user_id}")
71def get_user(user_id: int):
72    if user_id not in users:
73        raise HTTPException(status_code=404, detail="User not found")
74    return users[user_id]
75 
76@app.post("/users")
77def create_user(req: SignupRequest):
78    return signup(req)
79 
80@app.post("/users/{user_id}/make-manager")
81def make_manager(user_id: int, req: MakeManagerRequest, authorization: Optional[str] = Header(None)):
82    current_user_id = get_current_user(authorization)
83    if user_id not in users:
84        raise HTTPException(status_code=404, detail="User not found")
85    users[user_id]["is_manager"] = True
86    return {"message": f"User {user_id} is now a manager"}
87 
88@app.post("/orders")
89def create_order(req: OrderRequest, authorization: Optional[str] = Header(None)):
90    current_user_id = get_current_user(authorization)
91    if req.user_id not in users:
92        raise HTTPException(status_code=404, detail="User not found")
93    global next_order_id
94    order_id = next_order_id
95    next_order_id += 1
96    points_earned = int(req.amount)
97    orders[order_id] = {
98        "id": order_id,
99        "user_id": req.user_id,
100        "amount": req.amount,
101        "points_earned": points_earned
102    }
103    users[req.user_id]["points"] += points_earned
104    return {"order_id": order_id, "points_earned": points_earned}
105 
106@app.get("/orders/{order_id}")
107def get_order(order_id: int):
108    if order_id not in orders:
109        raise HTTPException(status_code=404, detail="Order not found")
110    return orders[order_id]
111 
112@app.post("/rewards")
113def create_reward(req: RewardRequest, authorization: Optional[str] = Header(None)):
114    current_user_id = get_current_user(authorization)
115    if not users[current_user_id]["is_manager"]:
116        raise HTTPException(status_code=403, detail="Only managers can create rewards")
117    global next_reward_id
118    reward_id = next_reward_id
119    next_reward_id += 1
120    rewards[reward_id] = {
121        "id": reward_id,
122        "name": req.name,
123        "points_required": req.points_required,
124        "description": req.description
125    }
126    return {"reward_id": reward_id, "message": "Reward created"}
127 
128@app.get("/rewards/{reward_id}")
129def get_reward(reward_id: int):
130    if reward_id not in rewards:
131        raise HTTPException(status_code=404, detail="Reward not found")
132    return rewards[reward_id]
133 
134@app.post("/redeem")
135def redeem_reward(user_id: int, reward_id: int, authorization: Optional[str] = Header(None)):
136    current_user_id = get_current_user(authorization)
137    if user_id not in users:
138        raise HTTPException(status_code=404, detail="User not found")
139    if reward_id not in rewards:
140        raise HTTPException(status_code=404, detail="Reward not found")
141    reward = rewards[reward_id]
142    user = users[user_id]
143    if user["points"] < reward["points_required"]:
144        raise HTTPException(status_code=400, detail="Not enough points")
145    user["points"] -= reward["points_required"]
146    return {"message": f"Redeemed {reward['name']}", "points_remaining": user["points"]}

1fastapi
2uvicorn
3pydantic