Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9orders = {}
10products = {}
11next_user_id = 1
12next_order_id = 1
13next_product_id = 1
14tokens = {}
15roles = {}
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class OrderCreate(BaseModel):
26    product: str
27    amount: float
28 
29class ProductCreate(BaseModel):
30    name: str
31    price: float
32 
33def get_current_user(authorization: Optional[str] = Header(None)):
34    if not authorization:
35        raise HTTPException(status_code=401, detail="No auth header")
36    token = authorization.replace("Bearer ", "")
37    if token not in tokens:
38        raise HTTPException(status_code=401, detail="Invalid token")
39    return tokens[token]
40 
41@app.post("/signup")
42def signup(req: SignupRequest):
43    global next_user_id
44    for u in users.values():
45        if u["username"] == req.username:
46            raise HTTPException(status_code=400, detail="Username exists")
47    user_id = next_user_id
48    next_user_id += 1
49    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
50    roles[user_id] = "user"
51    return {"user_id": user_id, "username": req.username}
52 
53@app.post("/login")
54def login(req: LoginRequest):
55    for uid, u in users.items():
56        if u["username"] == req.username and u["password"] == req.password:
57            token = secrets.token_hex(16)
58            tokens[token] = uid
59            return {"token": token}
60    raise HTTPException(status_code=401, detail="Invalid credentials")
61 
62@app.post("/promote/{user_id}")
63def promote(user_id: int, authorization: Optional[str] = Header(None)):
64    admin_id = get_current_user(authorization)
65    if roles.get(admin_id) != "admin":
66        raise HTTPException(status_code=403, detail="Not admin")
67    if user_id not in users:
68        raise HTTPException(status_code=404, detail="User not found")
69    roles[user_id] = "analyst"
70    return {"status": "promoted", "user_id": user_id, "role": "analyst"}
71 
72@app.get("/orders/{order_id}")
73def get_order(order_id: int, authorization: Optional[str] = Header(None)):
74    get_current_user(authorization)
75    if order_id not in orders:
76        raise HTTPException(status_code=404, detail="Order not found")
77    return orders[order_id]
78 
79@app.get("/products/{product_id}")
80def get_product(product_id: int, authorization: Optional[str] = Header(None)):
81    get_current_user(authorization)
82    if product_id not in products:
83        raise HTTPException(status_code=404, detail="Product not found")
84    return products[product_id]
85 
86@app.post("/orders")
87def create_order(order: OrderCreate, authorization: Optional[str] = Header(None)):
88    user_id = get_current_user(authorization)
89    global next_order_id
90    oid = next_order_id
91    next_order_id += 1
92    orders[oid] = {"id": oid, "user_id": user_id, "product": order.product, "amount": order.amount}
93    return orders[oid]
94 
95@app.post("/products")
96def create_product(product: ProductCreate, authorization: Optional[str] = Header(None)):
97    get_current_user(authorization)
98    global next_product_id
99    pid = next_product_id
100    next_product_id += 1
101    products[pid] = {"id": pid, "name": product.name, "price": product.price}
102    return products[pid]

1fastapi
2uvicorn