Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4import datetime
5 
6app = FastAPI()
7 
8users = {}
9couriers = {}
10deliveries = {}
11tokens = {}
12user_id_counter = 1
13courier_id_counter = 1
14delivery_id_counter = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19    role: str = "customer"
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class CourierCreate(BaseModel):
26    name: str
27    phone: str
28 
29class DeliveryCreate(BaseModel):
30    customer_id: int
31    pickup: str
32    dropoff: str
33    status: str = "pending"
34 
35class DeliveryUpdate(BaseModel):
36    status: str
37 
38def get_current_user(authorization: str = Header(None)):
39    if not authorization:
40        raise HTTPException(status_code=401, detail="No auth token")
41    token = authorization.replace("Bearer ", "")
42    if token not in tokens:
43        raise HTTPException(status_code=401, detail="Invalid token")
44    return tokens[token]
45 
46@app.post("/signup")
47def signup(req: SignupRequest):
48    global user_id_counter
49    for u in users.values():
50        if u["username"] == req.username:
51            raise HTTPException(status_code=400, detail="Username exists")
52    user_id = user_id_counter
53    user_id_counter += 1
54    users[user_id] = {"id": user_id, "username": req.username, "password": req.password, "role": req.role}
55    return {"id": user_id, "username": req.username, "role": req.role}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for u in users.values():
60        if u["username"] == req.username and u["password"] == req.password:
61            token = secrets.token_hex(16)
62            tokens[token] = u
63            return {"token": token}
64    raise HTTPException(status_code=401, detail="Invalid credentials")
65 
66@app.post("/couriers")
67def create_courier(req: CourierCreate, authorization: str = Header(None)):
68    user = get_current_user(authorization)
69    if user["role"] != "admin":
70        raise HTTPException(status_code=403, detail="Only admins can create couriers")
71    global courier_id_counter
72    courier_id = courier_id_counter
73    courier_id_counter += 1
74    couriers[courier_id] = {"id": courier_id, "name": req.name, "phone": req.phone, "deliveries": []}
75    return couriers[courier_id]
76 
77@app.get("/couriers/{courier_id}")
78def get_courier(courier_id: int, authorization: str = Header(None)):
79    get_current_user(authorization)
80    if courier_id not in couriers:
81        raise HTTPException(status_code=404, detail="Courier not found")
82    return couriers[courier_id]
83 
84@app.post("/deliveries")
85def create_delivery(req: DeliveryCreate, authorization: str = Header(None)):
86    user = get_current_user(authorization)
87    global delivery_id_counter
88    delivery_id = delivery_id_counter
89    delivery_id_counter += 1
90    deliveries[delivery_id] = {
91        "id": delivery_id,
92        "customer_id": req.customer_id,
93        "pickup": req.pickup,
94        "dropoff": req.dropoff,
95        "status": req.status,
96        "courier_id": None,
97        "updated_at": datetime.datetime.utcnow().isoformat()
98    }
99    return deliveries[delivery_id]
100 
101@app.get("/deliveries/{delivery_id}")
102def get_delivery(delivery_id: int, authorization: str = Header(None)):
103    user = get_current_user(authorization)
104    if delivery_id not in deliveries:
105        raise HTTPException(status_code=404, detail="Delivery not found")
106    return deliveries[delivery_id]
107 
108@app.put("/deliveries/{delivery_id}/status")
109def update_delivery_status(delivery_id: int, req: DeliveryUpdate, authorization: str = Header(None)):
110    user = get_current_user(authorization)
111    if delivery_id not in deliveries:
112        raise HTTPException(status_code=404, detail="Delivery not found")
113    if user["role"] not in ["courier", "admin"]:
114        raise HTTPException(status_code=403, detail="Only couriers or admins can update status")
115    deliveries[delivery_id]["status"] = req.status
116    deliveries[delivery_id]["updated_at"] = datetime.datetime.utcnow().isoformat()
117    return deliveries[delivery_id]

1fastapi
2uvicorn