Browse the data

1from fastapi import FastAPI, HTTPException, Header, status
2from pydantic import BaseModel
3from typing import Optional, List
4from datetime import datetime
5import secrets
6import copy
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12shipments = {}
13shipment_id_counter = 0
14user_id_counter = 0
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class ShipmentCreate(BaseModel):
25    origin: str
26    destination: str
27    weight: float
28    carrier_tier: str
29 
30class ShipmentUpdate(BaseModel):
31    origin: Optional[str] = None
32    destination: Optional[str] = None
33    weight: Optional[float] = None
34    delivery_status: Optional[str] = None
35    carrier_tier: Optional[str] = None
36 
37def get_current_user(authorization: str = Header(None)):
38    if not authorization:
39        raise HTTPException(status_code=401, detail="Missing auth header")
40    token = authorization.replace("Bearer ", "")
41    if token not in tokens:
42        raise HTTPException(status_code=401, detail="Invalid token")
43    return tokens[token]
44 
45@app.post("/signup")
46def signup(req: SignupRequest):
47    global user_id_counter
48    user_id_counter += 1
49    users[user_id_counter] = {"id": user_id_counter, "username": req.username, "password": req.password}
50    return {"id": user_id_counter, "username": req.username}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    for uid, u in users.items():
55        if u["username"] == req.username and u["password"] == req.password:
56            token = secrets.token_hex(16)
57            tokens[token] = uid
58            return {"token": token}
59    raise HTTPException(status_code=401, detail="Invalid credentials")
60 
61@app.get("/shipments/{shipment_id}")
62def get_shipment(shipment_id: int, authorization: str = Header(None)):
63    get_current_user(authorization)
64    if shipment_id not in shipments:
65        raise HTTPException(status_code=404, detail="Shipment not found")
66    return shipments[shipment_id]
67 
68@app.post("/shipments")
69def create_shipment(req: ShipmentCreate, authorization: str = Header(None)):
70    global shipment_id_counter
71    user_id = get_current_user(authorization)
72    shipment_id_counter += 1
73    shipment = {
74        "id": shipment_id_counter,
75        "origin": req.origin,
76        "destination": req.destination,
77        "weight": req.weight,
78        "delivery_status": "pending",
79        "carrier_tier": req.carrier_tier,
80        "history": [],
81        "created_by": user_id
82    }
83    shipments[shipment_id_counter] = shipment
84    return shipment
85 
86@app.patch("/shipments/{shipment_id}")
87def update_shipment(shipment_id: int, req: ShipmentUpdate, authorization: str = Header(None)):
88    get_current_user(authorization)
89    if shipment_id not in shipments:
90        raise HTTPException(status_code=404, detail="Shipment not found")
91    
92    s = shipments[shipment_id]
93    old = copy.deepcopy(s)
94    
95    if req.origin is not None:
96        s["origin"] = req.origin
97    if req.destination is not None:
98        s["destination"] = req.destination
99    if req.weight is not None:
100        s["weight"] = req.weight
101    if req.delivery_status is not None:
102        s["delivery_status"] = req.delivery_status
103    if req.carrier_tier is not None:
104        s["carrier_tier"] = req.carrier_tier
105    
106    change_record = {
107        "timestamp": datetime.utcnow().isoformat(),
108        "old": old,
109        "new": copy.deepcopy(s)
110    }
111    s["history"].append(change_record)
112    
113    return s

1fastapi
2uvicorn