Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import hashlib
5import random
6import string
7 
8app = FastAPI()
9 
10drivers: Dict[int, dict] = {}
11users: Dict[str, dict] = {}
12tokens: Dict[str, str] = {}
13next_driver_id = 1
14next_user_id = 1
15 
16def generate_token():
17    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
18 
19def get_current_user(authorization: str = Header(...)):
20    if not authorization.startswith("Bearer "):
21        raise HTTPException(status_code=401, detail="Invalid auth header")
22    token = authorization[7:]
23    if token not in tokens:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return tokens[token]
26 
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35class DriverCreate(BaseModel):
36    name: str
37    vehicle_info: str
38    availability_hours: str
39    zone_rank: Optional[int] = None
40    is_express_driver: Optional[bool] = False
41 
42class DriverUpdate(BaseModel):
43    name: Optional[str] = None
44    vehicle_info: Optional[str] = None
45    availability_hours: Optional[str] = None
46    zone_rank: Optional[int] = None
47    is_express_driver: Optional[bool] = None
48 
49class DriverResponse(BaseModel):
50    id: int
51    name: str
52    vehicle_info: str
53    availability_hours: str
54    zone_rank: Optional[int]
55    is_express_driver: bool
56 
57@app.post("/signup")
58def signup(req: SignupRequest):
59    global next_user_id
60    if req.username in users:
61        raise HTTPException(status_code=400, detail="Username already exists")
62    user_id = next_user_id
63    next_user_id += 1
64    hashed_password = hashlib.sha256(req.password.encode()).hexdigest()
65    users[req.username] = {"id": user_id, "password": hashed_password}
66    token = generate_token()
67    tokens[token] = req.username
68    return {"token": token, "user_id": user_id}
69 
70@app.post("/login")
71def login(req: LoginRequest):
72    if req.username not in users:
73        raise HTTPException(status_code=401, detail="Invalid credentials")
74    hashed_password = hashlib.sha256(req.password.encode()).hexdigest()
75    if users[req.username]["password"] != hashed_password:
76        raise HTTPException(status_code=401, detail="Invalid credentials")
77    token = generate_token()
78    tokens[token] = req.username
79    return {"token": token}
80 
81@app.get("/drivers/{driver_id}", response_model=DriverResponse)
82def get_driver(driver_id: int, authorization: str = Header(...)):
83    get_current_user(authorization)
84    if driver_id not in drivers:
85        raise HTTPException(status_code=404, detail="Driver not found")
86    return drivers[driver_id]
87 
88@app.post("/drivers", response_model=DriverResponse)
89def create_driver(driver: DriverCreate, authorization: str = Header(...)):
90    global next_driver_id
91    get_current_user(authorization)
92    driver_id = next_driver_id
93    next_driver_id += 1
94    drivers[driver_id] = {
95        "id": driver_id,
96        "name": driver.name,
97        "vehicle_info": driver.vehicle_info,
98        "availability_hours": driver.availability_hours,
99        "zone_rank": driver.zone_rank,
100        "is_express_driver": driver.is_express_driver or False
101    }
102    return drivers[driver_id]
103 
104@app.put("/drivers/{driver_id}", response_model=DriverResponse)
105def update_driver(driver_id: int, driver: DriverUpdate, authorization: str = Header(...)):
106    get_current_user(authorization)
107    if driver_id not in drivers:
108        raise HTTPException(status_code=404, detail="Driver not found")
109    existing = drivers[driver_id]
110    if driver.name is not None:
111        existing["name"] = driver.name
112    if driver.vehicle_info is not None:
113        existing["vehicle_info"] = driver.vehicle_info
114    if driver.availability_hours is not None:
115        existing["availability_hours"] = driver.availability_hours
116    if driver.zone_rank is not None:
117        existing["zone_rank"] = driver.zone_rank
118    if driver.is_express_driver is not None:
119        existing["is_express_driver"] = driver.is_express_driver
120    return existing

1fastapi
2uvicorn