Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import uuid
5import time
6 
7app = FastAPI()
8 
9users: Dict[int, dict] = {}
10tokens: Dict[str, int] = {}
11inventory: Dict[int, dict] = {}
12next_user_id = 1
13next_inventory_id = 1
14 
15def get_user_id_from_token(authorization: str = Header(None)):
16    if not authorization:
17        raise HTTPException(status_code=401, detail="Missing Authorization header")
18    token = authorization.replace("Bearer ", "")
19    user_id = tokens.get(token)
20    if user_id is None:
21        raise HTTPException(status_code=401, detail="Invalid token")
22    return user_id
23 
24class SignupRequest(BaseModel):
25    username: str
26    password: str
27 
28class LoginRequest(BaseModel):
29    username: str
30    password: str
31 
32class InventoryItemCreate(BaseModel):
33    sku: str
34    quantity_on_hand: int
35    reorder_threshold: int
36 
37class InventoryAdjustRequest(BaseModel):
38    sku: str
39    quantity_change: int
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 already 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    return {"user_id": user_id, "username": req.username}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    for u in users.values():
55        if u["username"] == req.username and u["password"] == req.password:
56            token = str(uuid.uuid4())
57            tokens[token] = u["id"]
58            return {"token": token}
59    raise HTTPException(status_code=401, detail="Invalid credentials")
60 
61@app.get("/inventory")
62def list_inventory(authorization: str = Header(None)):
63    get_user_id_from_token(authorization)
64    return [{"sku": v["sku"], "quantity_on_hand": v["quantity_on_hand"], "reorder_threshold": v["reorder_threshold"]} for v in inventory.values()]
65 
66@app.post("/inventory/adjust")
67def adjust_inventory(req: InventoryAdjustRequest, authorization: str = Header(None)):
68    get_user_id_from_token(authorization)
69    for item in inventory.values():
70        if item["sku"] == req.sku:
71            item["quantity_on_hand"] += req.quantity_change
72            if item["quantity_on_hand"] < 0:
73                item["quantity_on_hand"] = 0
74            return {"sku": req.sku, "new_quantity": item["quantity_on_hand"]}
75    raise HTTPException(status_code=404, detail="SKU not found")
76 
77@app.get("/alerts")
78def get_alerts(authorization: str = Header(None)):
79    get_user_id_from_token(authorization)
80    alerts = []
81    for item in inventory.values():
82        if item["quantity_on_hand"] < item["reorder_threshold"]:
83            alerts.append({
84                "sku": item["sku"],
85                "quantity_on_hand": item["quantity_on_hand"],
86                "reorder_threshold": item["reorder_threshold"]
87            })
88    return alerts
89 
90@app.post("/inventory")
91def create_inventory(item: InventoryItemCreate, authorization: str = Header(None)):
92    global next_inventory_id
93    get_user_id_from_token(authorization)
94    inv_id = next_inventory_id
95    next_inventory_id += 1
96    inventory[inv_id] = {
97        "id": inv_id,
98        "sku": item.sku,
99        "quantity_on_hand": item.quantity_on_hand,
100        "reorder_threshold": item.reorder_threshold
101    }
102    return {"id": inv_id, "sku": item.sku}
103 
104@app.get("/inventory/{item_id}")
105def get_inventory_item(item_id: int, authorization: str = Header(None)):
106    get_user_id_from_token(authorization)
107    item = inventory.get(item_id)
108    if not item:
109        raise HTTPException(status_code=404, detail="Item not found")
110    return item

1fastapi
2uvicorn