Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4import time
5 
6app = FastAPI()
7 
8users = {}
9orders = {}
10transactions = {}
11rewards = {}
12tokens = {}
13 
14class SignupRequest(BaseModel):
15    username: str
16    password: str
17 
18class LoginRequest(BaseModel):
19    username: str
20    password: str
21 
22class OrderCreate(BaseModel):
23    user_id: int
24    amount: float
25 
26class RewardCreate(BaseModel):
27    name: str
28    cost: int
29 
30class RedeemRequest(BaseModel):
31    user_id: int
32    reward_id: int
33 
34user_id_counter = 1
35order_id_counter = 1
36reward_id_counter = 1
37transaction_id_counter = 1
38 
39def get_current_user(token: str = Header(...)):
40    for uid, t in tokens.items():
41        if t == token:
42            return uid
43    raise HTTPException(status_code=401, detail="Invalid token")
44 
45@app.post("/signup")
46def signup(req: SignupRequest):
47    global user_id_counter
48    uid = user_id_counter
49    user_id_counter += 1
50    users[uid] = {"id": uid, "username": req.username, "password": req.password, "points": 0}
51    token = secrets.token_hex(16)
52    tokens[uid] = token
53    return {"user_id": uid, "token": token}
54 
55@app.post("/login")
56def login(req: LoginRequest):
57    for uid, u in users.items():
58        if u["username"] == req.username and u["password"] == req.password:
59            token = secrets.token_hex(16)
60            tokens[uid] = token
61            return {"user_id": uid, "token": token}
62    raise HTTPException(status_code=401, detail="Invalid credentials")
63 
64@app.get("/users/{user_id}")
65def get_user(user_id: int, token: str = Header(...)):
66    get_current_user(token)
67    if user_id not in users:
68        raise HTTPException(status_code=404, detail="User not found")
69    return users[user_id]
70 
71@app.get("/orders/{order_id}")
72def get_order(order_id: int, token: str = Header(...)):
73    get_current_user(token)
74    if order_id not in orders:
75        raise HTTPException(status_code=404, detail="Order not found")
76    return orders[order_id]
77 
78@app.get("/transactions/{transaction_id}")
79def get_transaction(transaction_id: int, token: str = Header(...)):
80    get_current_user(token)
81    if transaction_id not in transactions:
82        raise HTTPException(status_code=404, detail="Transaction not found")
83    return transactions[transaction_id]
84 
85@app.get("/rewards/{reward_id}")
86def get_reward(reward_id: int, token: str = Header(...)):
87    get_current_user(token)
88    if reward_id not in rewards:
89        raise HTTPException(status_code=404, detail="Reward not found")
90    return rewards[reward_id]
91 
92@app.post("/orders")
93def create_order(req: OrderCreate, token: str = Header(...)):
94    global order_id_counter, transaction_id_counter
95    uid = get_current_user(token)
96    if req.user_id not in users:
97        raise HTTPException(status_code=404, detail="User not found")
98    oid = order_id_counter
99    order_id_counter += 1
100    points_earned = int(req.amount)
101    users[req.user_id]["points"] += points_earned
102    tid = transaction_id_counter
103    transaction_id_counter += 1
104    transactions[tid] = {
105        "id": tid,
106        "user_id": req.user_id,
107        "type": "earn",
108        "points": points_earned,
109        "order_id": oid,
110        "timestamp": time.time()
111    }
112    orders[oid] = {
113        "id": oid,
114        "user_id": req.user_id,
115        "amount": req.amount,
116        "points_earned": points_earned,
117        "transaction_id": tid
118    }
119    return orders[oid]
120 
121@app.post("/rewards")
122def create_reward(req: RewardCreate, token: str = Header(...)):
123    global reward_id_counter
124    get_current_user(token)
125    rid = reward_id_counter
126    reward_id_counter += 1
127    rewards[rid] = {"id": rid, "name": req.name, "cost": req.cost}
128    return rewards[rid]
129 
130@app.post("/redeem")
131def redeem(req: RedeemRequest, token: str = Header(...)):
132    global transaction_id_counter
133    get_current_user(token)
134    if req.user_id not in users:
135        raise HTTPException(status_code=404, detail="User not found")
136    if req.reward_id not in rewards:
137        raise HTTPException(status_code=404, detail="Reward not found")
138    reward = rewards[req.reward_id]
139    user = users[req.user_id]
140    if user["points"] < reward["cost"]:
141        raise HTTPException(status_code=400, detail="Not enough points")
142    user["points"] -= reward["cost"]
143    tid = transaction_id_counter
144    transaction_id_counter += 1
145    transactions[tid] = {
146        "id": tid,
147        "user_id": req.user_id,
148        "type": "redeem",
149        "points": -reward["cost"],
150        "reward_id": req.reward_id,
151        "timestamp": time.time()
152    }
153    return {"user_id": req.user_id, "points_remaining": user["points"], "transaction_id": tid}
154 
155@app.get("/points/{user_id}")
156def get_points(user_id: int, token: str = Header(...)):
157    get_current_user(token)
158    if user_id not in users:
159        raise HTTPException(status_code=404, detail="User not found")
160    return {"user_id": user_id, "points": users[user_id]["points"]}

1fastapi
2uvicorn