Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4from datetime import datetime
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10inventory = {}
11stock_logs = []
12user_id_counter = 1
13inventory_id_counter = 1
14 
15class UserCreate(BaseModel):
16    username: str
17    password: str
18 
19class UserLogin(BaseModel):
20    username: str
21    password: str
22 
23class InventoryCreate(BaseModel):
24    sku: str
25    quantity_on_hand: int
26    reorder_threshold: int
27    supplier_name: str
28 
29class StockUpdate(BaseModel):
30    sku: str
31    quantity_change: int
32 
33def get_current_user(authorization: str = Header(None)):
34    if not authorization or not authorization.startswith("Bearer "):
35        raise HTTPException(status_code=401, detail="Invalid auth")
36    token = authorization.split(" ")[1]
37    for uid, t in tokens.items():
38        if t == token:
39            return uid
40    raise HTTPException(status_code=401, detail="Invalid token")
41 
42@app.post("/signup")
43def signup(user: UserCreate):
44    global user_id_counter
45    if user.username in [u["username"] for u in users.values()]:
46        raise HTTPException(status_code=400, detail="Username taken")
47    uid = user_id_counter
48    users[uid] = {"id": uid, "username": user.username, "password": user.password}
49    user_id_counter += 1
50    return {"id": uid, "username": user.username}
51 
52@app.post("/login")
53def login(login_data: UserLogin):
54    for uid, u in users.items():
55        if u["username"] == login_data.username and u["password"] == login_data.password:
56            token = secrets.token_hex(32)
57            tokens[uid] = token
58            return {"token": token}
59    raise HTTPException(status_code=401, detail="Invalid credentials")
60 
61@app.get("/inventory/{id}")
62def get_inventory_item(id: int, authorization: str = Header(None)):
63    get_current_user(authorization)
64    if id not in inventory:
65        raise HTTPException(status_code=404, detail="Not found")
66    return inventory[id]
67 
68@app.post("/inventory")
69def create_inventory(item: InventoryCreate, authorization: str = Header(None)):
70    global inventory_id_counter
71    get_current_user(authorization)
72    new_id = inventory_id_counter
73    inventory[new_id] = {
74        "id": new_id,
75        "sku": item.sku,
76        "quantity_on_hand": item.quantity_on_hand,
77        "reorder_threshold": item.reorder_threshold,
78        "supplier_name": item.supplier_name
79    }
80    inventory_id_counter += 1
81    return inventory[new_id]
82 
83@app.get("/inventory")
84def get_all_inventory(authorization: str = Header(None)):
85    get_current_user(authorization)
86    return list(inventory.values())
87 
88@app.post("/stock-update")
89def stock_update(update: StockUpdate, authorization: str = Header(None)):
90    get_current_user(authorization)
91    for item_id, item in inventory.items():
92        if item["sku"] == update.sku:
93            old_qty = item["quantity_on_hand"]
94            item["quantity_on_hand"] += update.quantity_change
95            stock_logs.append({
96                "sku": update.sku,
97                "old_quantity": old_qty,
98                "new_quantity": item["quantity_on_hand"],
99                "change": update.quantity_change,
100                "timestamp": datetime.now().isoformat()
101            })
102            return {"message": "Updated", "item": item}
103    raise HTTPException(status_code=404, detail="SKU not found")

1fastapi
2uvicorn