Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import random
4import string
5import hashlib
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11shipments = {}
12shipment_id_counter = 1
13user_id_counter = 1
14 
15def generate_token():
16    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
17 
18def get_user_from_token(authorization: str = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing authorization header")
21    token = authorization.replace("Bearer ", "")
22    if token not in tokens:
23        raise HTTPException(status_code=401, detail="Invalid token")
24    return tokens[token]
25 
26@app.post("/signup")
27def signup(username: str, password: str):
28    global user_id_counter
29    if any(u["username"] == username for u in users.values()):
30        raise HTTPException(status_code=400, detail="Username already exists")
31    user_id = user_id_counter
32    user_id_counter += 1
33    users[user_id] = {
34        "id": user_id,
35        "username": username,
36        "password": hashlib.sha256(password.encode()).hexdigest()
37    }
38    token = generate_token()
39    tokens[token] = user_id
40    return {"user_id": user_id, "token": token}
41 
42@app.post("/login")
43def login(username: str, password: str):
44    for uid, user in users.items():
45        if user["username"] == username and user["password"] == hashlib.sha256(password.encode()).hexdigest():
46            token = generate_token()
47            tokens[token] = uid
48            return {"token": token}
49    raise HTTPException(status_code=401, detail="Invalid credentials")
50 
51@app.get("/shipments")
52def get_shipments(authorization: str = Header(None)):
53    get_user_from_token(authorization)
54    active = []
55    for sid, s in shipments.items():
56        if s["status"] != "delivered":
57            active.append({
58                "id": sid,
59                "origin": s["origin"],
60                "destination": s["destination"],
61                "status": s["status"],
62                "estimated_delivery": s["estimated_delivery"]
63            })
64    return active
65 
66@app.get("/shipments/{shipment_id}")
67def get_shipment(shipment_id: int, authorization: str = Header(None)):
68    get_user_from_token(authorization)
69    if shipment_id not in shipments:
70        raise HTTPException(status_code=404, detail="Shipment not found")
71    return shipments[shipment_id]
72 
73@app.post("/shipments")
74def create_shipment(origin: str, destination: str, estimated_delivery: str, authorization: str = Header(None)):
75    global shipment_id_counter
76    get_user_from_token(authorization)
77    shipment_id = shipment_id_counter
78    shipment_id_counter += 1
79    shipments[shipment_id] = {
80        "id": shipment_id,
81        "origin": origin,
82        "destination": destination,
83        "status": "in_transit",
84        "estimated_delivery": estimated_delivery,
85        "gps_coordinates": None
86    }
87    return {"id": shipment_id}
88 
89@app.get("/shipments/{shipment_id}/track")
90def track_shipment(shipment_id: int, authorization: str = Header(None)):
91    get_user_from_token(authorization)
92    if shipment_id not in shipments:
93        raise HTTPException(status_code=404, detail="Shipment not found")
94    s = shipments[shipment_id]
95    if s["gps_coordinates"]:
96        return {"gps_coordinates": s["gps_coordinates"]}
97    return {"gps_coordinates": None, "message": "GPS coordinates not available"}

1fastapi
2uvicorn