Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9products = {}
10suppliers = {}
11orders = {}
12next_user_id = 1
13next_product_id = 1
14next_supplier_id = 1
15next_order_id = 1
16reorder_threshold = {}
17 
18def get_current_user(authorization: Optional[str] = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing Authorization header")
21    token = authorization.replace("Bearer ", "")
22    if token not in tokens:
23        raise HTTPException(status_code=401, detail="Invalid token")
24    return tokens[token]
25 
26@app.post("/signup")
27def signup(username: str, password: str):
28    global next_user_id
29    for u in users.values():
30        if u["username"] == username:
31            raise HTTPException(status_code=400, detail="Username already exists")
32    user_id = next_user_id
33    next_user_id += 1
34    users[user_id] = {"id": user_id, "username": username, "password": password}
35    return {"id": user_id, "username": username}
36 
37@app.post("/login")
38def login(username: str, password: str):
39    for u in users.values():
40        if u["username"] == username and u["password"] == password:
41            token = secrets.token_hex(16)
42            tokens[token] = u["id"]
43            return {"token": token}
44    raise HTTPException(status_code=401, detail="Invalid credentials")
45 
46@app.get("/products/{product_id}")
47def get_product(product_id: int, authorization: Optional[str] = Header(None)):
48    get_current_user(authorization)
49    if product_id not in products:
50        raise HTTPException(status_code=404, detail="Product not found")
51    return products[product_id]
52 
53@app.post("/products")
54def create_product(name: str, stock: int = 0, threshold: int = 0, authorization: Optional[str] = Header(None)):
55    get_current_user(authorization)
56    global next_product_id
57    product_id = next_product_id
58    next_product_id += 1
59    products[product_id] = {"id": product_id, "name": name, "stock": stock}
60    if threshold > 0:
61        reorder_threshold[product_id] = threshold
62    if stock <= threshold:
63        print(f"ALERT: Restock needed for product {product_id} ({name})")
64    return products[product_id]
65 
66@app.get("/suppliers/{supplier_id}")
67def get_supplier(supplier_id: int, authorization: Optional[str] = Header(None)):
68    get_current_user(authorization)
69    if supplier_id not in suppliers:
70        raise HTTPException(status_code=404, detail="Supplier not found")
71    return suppliers[supplier_id]
72 
73@app.post("/suppliers")
74def create_supplier(name: str, authorization: Optional[str] = Header(None)):
75    get_current_user(authorization)
76    global next_supplier_id
77    supplier_id = next_supplier_id
78    next_supplier_id += 1
79    suppliers[supplier_id] = {"id": supplier_id, "name": name}
80    return suppliers[supplier_id]
81 
82@app.post("/suppliers/{supplier_id}/update-stock/{product_id}")
83def supplier_update_stock(supplier_id: int, product_id: int, quantity: int, authorization: Optional[str] = Header(None)):
84    get_current_user(authorization)
85    if supplier_id not in suppliers:
86        raise HTTPException(status_code=404, detail="Supplier not found")
87    if product_id not in products:
88        raise HTTPException(status_code=404, detail="Product not found")
89    products[product_id]["stock"] += quantity
90    threshold = reorder_threshold.get(product_id, 0)
91    if products[product_id]["stock"] <= threshold:
92        print(f"ALERT: Restock needed for product {product_id} ({products[product_id]['name']})")
93    return products[product_id]
94 
95@app.get("/orders/{order_id}")
96def get_order(order_id: int, authorization: Optional[str] = Header(None)):
97    get_current_user(authorization)
98    if order_id not in orders:
99        raise HTTPException(status_code=404, detail="Order not found")
100    return orders[order_id]
101 
102@app.post("/orders")
103def create_order(product_id: int, quantity: int, authorization: Optional[str] = Header(None)):
104    get_current_user(authorization)
105    if product_id not in products:
106        raise HTTPException(status_code=404, detail="Product not found")
107    global next_order_id
108    order_id = next_order_id
109    next_order_id += 1
110    orders[order_id] = {"id": order_id, "product_id": product_id, "quantity": quantity, "status": "pending"}
111    products[product_id]["stock"] -= quantity
112    threshold = reorder_threshold.get(product_id, 0)
113    if products[product_id]["stock"] <= threshold:
114        print(f"ALERT: Restock needed for product {product_id} ({products[product_id]['name']})")
115    return orders[order_id]

1fastapi
2uvicorn