Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict, List
4import secrets
5 
6app = FastAPI()
7 
8users: Dict[int, dict] = {}
9tokens: Dict[str, int] = {}
10next_user_id = 1
11 
12warehouses: Dict[int, dict] = {}
13next_warehouse_id = 1
14 
15stock_items: Dict[int, dict] = {}
16next_stock_item_id = 1
17 
18purchase_orders: Dict[int, dict] = {}
19next_po_id = 1
20 
21class SignupRequest(BaseModel):
22    username: str
23    password: str
24 
25class LoginRequest(BaseModel):
26    username: str
27    password: str
28 
29class WarehouseCreate(BaseModel):
30    name: str
31 
32class StockItemCreate(BaseModel):
33    warehouse_id: int
34    name: str
35    bin_location: str
36    quantity: int
37    reorder_threshold: int
38 
39class RestockRequest(BaseModel):
40    warehouse_id: int
41 
42def auth_user(authorization: Optional[str] = Header(None)) -> int:
43    if not authorization:
44        raise HTTPException(status_code=401, detail="Missing auth header")
45    token = authorization.replace("Bearer ", "")
46    if token not in tokens:
47        raise HTTPException(status_code=401, detail="Invalid token")
48    return tokens[token]
49 
50@app.post("/signup")
51def signup(req: SignupRequest):
52    global next_user_id
53    uid = next_user_id
54    next_user_id += 1
55    users[uid] = {"username": req.username, "password": req.password}
56    token = secrets.token_hex(16)
57    tokens[token] = uid
58    return {"user_id": uid, "token": token}
59 
60@app.post("/login")
61def login(req: LoginRequest):
62    for uid, u in users.items():
63        if u["username"] == req.username and u["password"] == req.password:
64            token = secrets.token_hex(16)
65            tokens[token] = uid
66            return {"user_id": uid, "token": token}
67    raise HTTPException(status_code=401, detail="Invalid credentials")
68 
69@app.post("/warehouse")
70def create_warehouse(req: WarehouseCreate, authorization: Optional[str] = Header(None)):
71    auth_user(authorization)
72    global next_warehouse_id
73    wid = next_warehouse_id
74    next_warehouse_id += 1
75    warehouses[wid] = {"id": wid, "name": req.name}
76    return warehouses[wid]
77 
78@app.get("/warehouse/{wid}")
79def get_warehouse(wid: int, authorization: Optional[str] = Header(None)):
80    auth_user(authorization)
81    if wid not in warehouses:
82        raise HTTPException(status_code=404, detail="Warehouse not found")
83    return warehouses[wid]
84 
85@app.get("/warehouse/{wid}/stock")
86def get_stock(wid: int, authorization: Optional[str] = Header(None)):
87    auth_user(authorization)
88    if wid not in warehouses:
89        raise HTTPException(status_code=404, detail="Warehouse not found")
90    items = []
91    for sid, item in stock_items.items():
92        if item["warehouse_id"] == wid:
93            reorder_alert = item["quantity"] <= item["reorder_threshold"]
94            items.append({
95                "name": item["name"],
96                "bin_location": item["bin_location"],
97                "quantity": item["quantity"],
98                "reorder_alert": reorder_alert
99            })
100    return {"warehouse_id": wid, "stock": items}
101 
102@app.post("/stock")
103def create_stock_item(req: StockItemCreate, authorization: Optional[str] = Header(None)):
104    auth_user(authorization)
105    global next_stock_item_id
106    sid = next_stock_item_id
107    next_stock_item_id += 1
108    stock_items[sid] = {
109        "id": sid,
110        "warehouse_id": req.warehouse_id,
111        "name": req.name,
112        "bin_location": req.bin_location,
113        "quantity": req.quantity,
114        "reorder_threshold": req.reorder_threshold
115    }
116    return stock_items[sid]
117 
118@app.get("/stock/{sid}")
119def get_stock_item(sid: int, authorization: Optional[str] = Header(None)):
120    auth_user(authorization)
121    if sid not in stock_items:
122        raise HTTPException(status_code=404, detail="Stock item not found")
123    return stock_items[sid]
124 
125@app.post("/warehouse/restock")
126def restock_warehouse(req: RestockRequest, authorization: Optional[str] = Header(None)):
127    auth_user(authorization)
128    wid = req.warehouse_id
129    if wid not in warehouses:
130        raise HTTPException(status_code=404, detail="Warehouse not found")
131    low_stock_items = []
132    for sid, item in stock_items.items():
133        if item["warehouse_id"] == wid and item["quantity"] <= item["reorder_threshold"]:
134            low_stock_items.append(item)
135    if not low_stock_items:
136        return {"message": "No low stock items", "purchase_order_id": None}
137    global next_po_id
138    po_id = next_po_id
139    next_po_id += 1
140    po_items = []
141    for item in low_stock_items:
142        order_qty = item["reorder_threshold"] * 2 - item["quantity"]
143        po_items.append({
144            "stock_item_id": item["id"],
145            "name": item["name"],
146            "order_quantity": order_qty
147        })
148    purchase_orders[po_id] = {
149        "id": po_id,
150        "warehouse_id": wid,
151        "items": po_items
152    }
153    return {"message": "Purchase order created", "purchase_order_id": po_id, "items": po_items}
154 
155@app.get("/purchase_order/{po_id}")
156def get_purchase_order(po_id: int, authorization: Optional[str] = Header(None)):
157    auth_user(authorization)
158    if po_id not in purchase_orders:
159        raise HTTPException(status_code=404, detail="Purchase order not found")
160    return purchase_orders[po_id]

1fastapi
2uvicorn