Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import hashlib
5import secrets
6 
7app = FastAPI()
8 
9users = {}
10drivers = {}
11tokens = {}
12user_id_counter = 1
13driver_id_counter = 1
14 
15class SignupRequest(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class DriverCreate(BaseModel):
24    name: str
25    license_info: str
26    delivery_zone: str
27    dispatch_priority: int
28 
29class DriverUpdate(BaseModel):
30    name: Optional[str] = None
31    license_info: Optional[str] = None
32    delivery_zone: Optional[str] = None
33    dispatch_priority: Optional[int] = None
34 
35def hash_password(password: str) -> str:
36    return hashlib.sha256(password.encode()).hexdigest()
37 
38def get_user_from_token(authorization: str = Header(None)):
39    if not authorization:
40        raise HTTPException(status_code=401, detail="No auth header")
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    if req.username in users:
50        raise HTTPException(status_code=400, detail="Username already exists")
51    user_id = user_id_counter
52    user_id_counter += 1
53    users[req.username] = {"id": user_id, "password": hash_password(req.password)}
54    return {"id": user_id, "username": req.username}
55 
56@app.post("/login")
57def login(req: LoginRequest):
58    if req.username not in users:
59        raise HTTPException(status_code=401, detail="Invalid credentials")
60    if users[req.username]["password"] != hash_password(req.password):
61        raise HTTPException(status_code=401, detail="Invalid credentials")
62    token = secrets.token_hex(32)
63    tokens[token] = req.username
64    return {"token": token}
65 
66@app.get("/drivers/{driver_id}")
67def get_driver(driver_id: int, authorization: str = Header(None)):
68    get_user_from_token(authorization)
69    if driver_id not in drivers:
70        raise HTTPException(status_code=404, detail="Driver not found")
71    return drivers[driver_id]
72 
73@app.post("/drivers")
74def create_driver(driver: DriverCreate, authorization: str = Header(None)):
75    global driver_id_counter
76    get_user_from_token(authorization)
77    driver_id = driver_id_counter
78    driver_id_counter += 1
79    drivers[driver_id] = {
80        "id": driver_id,
81        "name": driver.name,
82        "license_info": driver.license_info,
83        "delivery_zone": driver.delivery_zone,
84        "dispatch_priority": driver.dispatch_priority
85    }
86    return drivers[driver_id]
87 
88@app.put("/drivers/{driver_id}")
89def update_driver(driver_id: int, driver: DriverUpdate, authorization: str = Header(None)):
90    get_user_from_token(authorization)
91    if driver_id not in drivers:
92        raise HTTPException(status_code=404, detail="Driver not found")
93    if driver.name is not None:
94        drivers[driver_id]["name"] = driver.name
95    if driver.license_info is not None:
96        drivers[driver_id]["license_info"] = driver.license_info
97    if driver.delivery_zone is not None:
98        drivers[driver_id]["delivery_zone"] = driver.delivery_zone
99    if driver.dispatch_priority is not None:
100        drivers[driver_id]["dispatch_priority"] = driver.dispatch_priority
101    return drivers[driver_id]

1fastapi
2uvicorn