Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import uvicorn
6 
7app = FastAPI()
8 
9users = {}
10orders = {}
11packages = {}
12shipments = {}
13dispatchers = {}
14managers = {}
15tokens = {}
16next_user_id = 1
17next_order_id = 1
18next_package_id = 1
19next_shipment_id = 1
20next_dispatcher_id = 1
21next_manager_id = 1
22 
23class SignupRequest(BaseModel):
24    username: str
25    password: str
26    role: str = "user"
27 
28class LoginRequest(BaseModel):
29    username: str
30    password: str
31 
32class OrderCreate(BaseModel):
33    description: str
34    destination: str
35 
36class PackageCreate(BaseModel):
37    weight: float
38    dimensions: str
39 
40class ShipmentCreate(BaseModel):
41    origin: str
42    destination: str
43 
44def get_current_user(authorization: Optional[str] = Header(None)):
45    if not authorization:
46        raise HTTPException(status_code=401, detail="No auth header")
47    token = authorization.replace("Bearer ", "")
48    if token not in tokens:
49        raise HTTPException(status_code=401, detail="Invalid token")
50    return tokens[token]
51 
52@app.post("/signup")
53def signup(req: SignupRequest):
54    global next_user_id
55    user_id = next_user_id
56    next_user_id += 1
57    users[user_id] = {"id": user_id, "username": req.username, "password": req.password, "role": req.role}
58    if req.role == "dispatcher":
59        dispatchers[user_id] = users[user_id]
60    return {"user_id": user_id, "username": req.username, "role": req.role}
61 
62@app.post("/login")
63def login(req: LoginRequest):
64    for uid, u in users.items():
65        if u["username"] == req.username and u["password"] == req.password:
66            token = secrets.token_hex(16)
67            tokens[token] = uid
68            return {"token": token}
69    raise HTTPException(status_code=401, detail="Invalid credentials")
70 
71@app.post("/promote/{user_id}")
72def promote(user_id: int, authorization: Optional[str] = Header(None)):
73    admin_id = get_current_user(authorization)
74    if users[admin_id]["role"] != "admin":
75        raise HTTPException(status_code=403, detail="Only admins can promote")
76    if user_id not in users:
77        raise HTTPException(status_code=404, detail="User not found")
78    user = users[user_id]
79    if user["role"] != "dispatcher":
80        raise HTTPException(status_code=400, detail="Only dispatchers can be promoted to manager")
81    user["role"] = "manager"
82    if user_id in dispatchers:
83        del dispatchers[user_id]
84    managers[user_id] = user
85    return {"message": f"User {user_id} promoted to manager"}
86 
87@app.get("/users/{user_id}")
88def get_user(user_id: int):
89    if user_id not in users:
90        raise HTTPException(status_code=404, detail="User not found")
91    return users[user_id]
92 
93@app.post("/orders")
94def create_order(order: OrderCreate, authorization: Optional[str] = Header(None)):
95    user_id = get_current_user(authorization)
96    global next_order_id
97    oid = next_order_id
98    next_order_id += 1
99    orders[oid] = {"id": oid, "description": order.description, "destination": order.destination, "user_id": user_id}
100    return orders[oid]
101 
102@app.get("/orders/{order_id}")
103def get_order(order_id: int):
104    if order_id not in orders:
105        raise HTTPException(status_code=404, detail="Order not found")
106    return orders[order_id]
107 
108@app.post("/packages")
109def create_package(pkg: PackageCreate, authorization: Optional[str] = Header(None)):
110    user_id = get_current_user(authorization)
111    global next_package_id
112    pid = next_package_id
113    next_package_id += 1
114    packages[pid] = {"id": pid, "weight": pkg.weight, "dimensions": pkg.dimensions, "user_id": user_id}
115    return packages[pid]
116 
117@app.get("/packages/{package_id}")
118def get_package(package_id: int):
119    if package_id not in packages:
120        raise HTTPException(status_code=404, detail="Package not found")
121    return packages[package_id]
122 
123@app.post("/shipments")
124def create_shipment(ship: ShipmentCreate, authorization: Optional[str] = Header(None)):
125    user_id = get_current_user(authorization)
126    global next_shipment_id
127    sid = next_shipment_id
128    next_shipment_id += 1
129    shipments[sid] = {"id": sid, "origin": ship.origin, "destination": ship.destination, "user_id": user_id}
130    return shipments[sid]
131 
132@app.get("/shipments/{shipment_id}")
133def get_shipment(shipment_id: int):
134    if shipment_id not in shipments:
135        raise HTTPException(status_code=404, detail="Shipment not found")
136    return shipments[shipment_id]

1fastapi
2uvicorn