Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4import uvicorn
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10deliveries = {}
11next_user_id = 1
12next_delivery_id = 1
13next_token_id = 1
14 
15class SignupRequest(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class DeliveryCreate(BaseModel):
24    status: str = "pending"
25    description: str = ""
26 
27def get_user_id_from_token(authorization: str = Header(None)):
28    if not authorization:
29        raise HTTPException(status_code=401, detail="Missing Authorization header")
30    token = authorization.replace("Bearer ", "")
31    user_id = tokens.get(token)
32    if user_id is None:
33        raise HTTPException(status_code=401, detail="Invalid token")
34    return user_id
35 
36@app.post("/signup")
37def signup(req: SignupRequest):
38    global next_user_id
39    user_id = next_user_id
40    next_user_id += 1
41    users[user_id] = {"username": req.username, "password": req.password}
42    return {"user_id": user_id, "message": "User created"}
43 
44@app.post("/login")
45def login(req: LoginRequest):
46    for uid, u in users.items():
47        if u["username"] == req.username and u["password"] == req.password:
48            token = secrets.token_hex(16)
49            tokens[token] = uid
50            return {"token": token}
51    raise HTTPException(status_code=401, detail="Invalid credentials")
52 
53@app.get("/delivery/{delivery_id}")
54def get_delivery(delivery_id: int, authorization: str = Header(None)):
55    user_id = get_user_id_from_token(authorization)
56    delivery = deliveries.get(delivery_id)
57    if not delivery:
58        raise HTTPException(status_code=404, detail="Delivery not found")
59    return delivery
60 
61@app.post("/delivery")
62def create_delivery(req: DeliveryCreate, authorization: str = Header(None)):
63    global next_delivery_id
64    user_id = get_user_id_from_token(authorization)
65    delivery_id = next_delivery_id
66    next_delivery_id += 1
67    deliveries[delivery_id] = {
68        "id": delivery_id,
69        "status": req.status,
70        "description": req.description,
71        "created_by": user_id
72    }
73    return deliveries[delivery_id]
74 
75@app.post("/courier/update/{delivery_id}")
76def update_delivery_status(delivery_id: int, status: str, authorization: str = Header(None)):
77    user_id = get_user_id_from_token(authorization)
78    delivery = deliveries.get(delivery_id)
79    if not delivery:
80        raise HTTPException(status_code=404, detail="Delivery not found")
81    delivery["status"] = status
82    return delivery
83 
84if __name__ == "__main__":
85    uvicorn.run(app, host="127.0.0.1", port=8000)

1fastapi
2uvicorn