Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import random
5import string
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11items = {}
12restock_history = {}
13item_id_counter = 0
14user_id_counter = 0
15 
16def generate_token():
17    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
18 
19def get_current_user(authorization: str = Header(None)):
20    if not authorization:
21        raise HTTPException(status_code=401, detail="No auth token")
22    token = authorization.replace("Bearer ", "")
23    user_id = tokens.get(token)
24    if not user_id:
25        raise HTTPException(status_code=401, detail="Invalid token")
26    return user_id
27 
28class SignupRequest(BaseModel):
29    username: str
30    password: str
31 
32class LoginRequest(BaseModel):
33    username: str
34    password: str
35 
36class ItemCreate(BaseModel):
37    name: str
38    quantity: int
39    location: str
40    reorder_threshold: int
41 
42class ItemUpdate(BaseModel):
43    quantity: Optional[int] = None
44    location: Optional[str] = None
45    reorder_threshold: Optional[int] = None
46 
47class BatchUpdateItem(BaseModel):
48    id: int
49    quantity: Optional[int] = None
50    location: Optional[str] = None
51    reorder_threshold: Optional[int] = None
52 
53class BatchUpdateRequest(BaseModel):
54    updates: List[BatchUpdateItem]
55 
56@app.post("/signup")
57def signup(req: SignupRequest):
58    global user_id_counter
59    user_id_counter += 1
60    users[user_id_counter] = {"username": req.username, "password": req.password}
61    token = generate_token()
62    tokens[token] = user_id_counter
63    return {"user_id": user_id_counter, "token": token}
64 
65@app.post("/login")
66def login(req: LoginRequest):
67    for uid, u in users.items():
68        if u["username"] == req.username and u["password"] == req.password:
69            token = generate_token()
70            tokens[token] = uid
71            return {"token": token}
72    raise HTTPException(status_code=401, detail="Invalid credentials")
73 
74@app.get("/items/{item_id}")
75def get_item(item_id: int, authorization: str = Header(None)):
76    get_current_user(authorization)
77    item = items.get(item_id)
78    if not item:
79        raise HTTPException(status_code=404, detail="Item not found")
80    return item
81 
82@app.post("/items")
83def create_item(item: ItemCreate, authorization: str = Header(None)):
84    global item_id_counter
85    user_id = get_current_user(authorization)
86    item_id_counter += 1
87    items[item_id_counter] = {
88        "id": item_id_counter,
89        "name": item.name,
90        "quantity": item.quantity,
91        "location": item.location,
92        "reorder_threshold": item.reorder_threshold,
93        "user_id": user_id
94    }
95    restock_history[item_id_counter] = []
96    return items[item_id_counter]
97 
98@app.patch("/items/{item_id}")
99def update_item(item_id: int, update: ItemUpdate, authorization: str = Header(None)):
100    get_current_user(authorization)
101    if item_id not in items:
102        raise HTTPException(status_code=404, detail="Item not found")
103    item = items[item_id]
104    old_quantity = item["quantity"]
105    if update.quantity is not None:
106        item["quantity"] = update.quantity
107    if update.location is not None:
108        item["location"] = update.location
109    if update.reorder_threshold is not None:
110        item["reorder_threshold"] = update.reorder_threshold
111    if update.quantity is not None and update.quantity > old_quantity:
112        restock_history[item_id].append({
113            "old_quantity": old_quantity,
114            "new_quantity": update.quantity,
115            "added": update.quantity - old_quantity
116        })
117    return item
118 
119@app.patch("/items/batch")
120def batch_update_items(batch: BatchUpdateRequest, authorization: str = Header(None)):
121    get_current_user(authorization)
122    results = []
123    for upd in batch.updates:
124        if upd.id not in items:
125            results.append({"id": upd.id, "status": "not_found"})
126            continue
127        item = items[upd.id]
128        old_quantity = item["quantity"]
129        if upd.quantity is not None:
130            item["quantity"] = upd.quantity
131        if upd.location is not None:
132            item["location"] = upd.location
133        if upd.reorder_threshold is not None:
134            item["reorder_threshold"] = upd.reorder_threshold
135        if upd.quantity is not None and upd.quantity > old_quantity:
136            restock_history[upd.id].append({
137                "old_quantity": old_quantity,
138                "new_quantity": upd.quantity,
139                "added": upd.quantity - old_quantity
140            })
141        results.append({"id": upd.id, "status": "updated"})
142    return {"results": results}
143 
144@app.get("/items/{item_id}/restock-history")
145def get_restock_history(item_id: int, authorization: str = Header(None)):
146    get_current_user(authorization)
147    if item_id not in items:
148        raise HTTPException(status_code=404, detail="Item not found")
149    return restock_history.get(item_id, [])

1fastapi
2uvicorn
3pydantic