Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime, timedelta
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_current_user(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 username in [u["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] = {"id": user_id, "username": username, "password": hashlib.sha256(password.encode()).hexdigest()}
34    return {"id": user_id, "username": username}
35 
36@app.post("/login")
37def login(username: str, password: str):
38    user = None
39    for u in users.values():
40        if u["username"] == username and u["password"] == hashlib.sha256(password.encode()).hexdigest():
41            user = u
42            break
43    if not user:
44        raise HTTPException(status_code=401, detail="Invalid credentials")
45    token = generate_token()
46    tokens[token] = user
47    return {"token": token}
48 
49@app.get("/shipments/{shipment_id}")
50def get_shipment(shipment_id: int, authorization: str = Header(None)):
51    get_current_user(authorization)
52    if shipment_id not in shipments:
53        raise HTTPException(status_code=404, detail="Shipment not found")
54    return shipments[shipment_id]
55 
56@app.post("/shipments")
57def create_shipment(destination: str, status: str = "pending", delivery_window: str = "morning", authorization: str = Header(None)):
58    global shipment_id_counter
59    get_current_user(authorization)
60    shipment_id = shipment_id_counter
61    shipment_id_counter += 1
62    eta_minutes = random.randint(30, 180)
63    driver_name = random.choice(["Alice", "Bob", "Charlie", "Diana", "Eve"])
64    shipments[shipment_id] = {
65        "id": shipment_id,
66        "destination": destination,
67        "status": status,
68        "delivery_window": delivery_window,
69        "eta_minutes": eta_minutes,
70        "driver_name": driver_name
71    }
72    return shipments[shipment_id]
73 
74@app.get("/shipments")
75def list_shipments(
76    status: str = None,
77    destination: str = None,
78    delivery_window: str = None,
79    authorization: str = Header(None)
80):
81    get_current_user(authorization)
82    result = []
83    for s in shipments.values():
84        if status and s["status"] != status:
85            continue
86        if destination and s["destination"] != destination:
87            continue
88        if delivery_window and s["delivery_window"] != delivery_window:
89            continue
90        result.append(s)
91    return result
92 
93@app.get("/{resource}/{resource_id}")
94def get_resource(resource: str, resource_id: int, authorization: str = Header(None)):
95    get_current_user(authorization)
96    store = {"users": users, "shipments": shipments}
97    if resource not in store:
98        raise HTTPException(status_code=404, detail="Resource not found")
99    data = store[resource]
100    if resource_id not in data:
101        raise HTTPException(status_code=404, detail="Resource not found")
102    return data[resource_id]
103 
104@app.post("/{resource}")
105def create_resource(resource: str, authorization: str = Header(None)):
106    get_current_user(authorization)
107    raise HTTPException(status_code=400, detail="Use specific endpoints like /shipments or /signup")

1fastapi
2uvicorn